Agents et outils LangChain : Construire des IA autonomes capables d'agir

1. Thought (Pensée) : "Je dois chercher des informations sur X"
2. Action (Action) : Utiliser l'outil "Search"
3. Observation : Résultat de la recherche
4. Thought : "J'ai trouvé Y, maintenant je dois calculer Z"
5. Action : Utiliser l'outil "Calculator"
6. Observation : Résultat du calcul
7. Thought : "J'ai toutes les informations, je peux répondre"
8. Final Answer : Réponse complète

User: Quelle est la racine carrée de l'âge du président français actuel ?

Thought: Je dois d'abord trouver qui est le président et son âge
Action: Search
Action Input: "président français 2025 âge"
Observation: Emmanuel Macron, né en 1977, a 48 ans en 2025

Thought: Maintenant je dois calculer la racine carrée de 48
Action: Calculator
Action Input: sqrt(48)
Observation: 6.928

Thought: J'ai la réponse finale
Final Answer: La racine carrée de l'âge d'Emmanuel Macron (48 ans) est environ 6.93.

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool
from langchain.prompts import PromptTemplate

# 1. LLM (le cerveau)
llm = ChatOpenAI(model="gpt-4", temperature=0)

# 2. Définition des outils
def search_web(query: str) -> str:
    """Recherche sur le web"""
    # Implémentation réelle avec une API de recherche
    return f"Résultats pour '{query}': ..."

def calculate(expression: str) -> str:
    """Calcule une expression mathématique de manière sécurisée"""
    import ast
    import operator

    # Opérateurs autorisés
    operators = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
        ast.Pow: operator.pow,
        ast.USub: operator.neg,
    }

    def eval_expr(node):
        if isinstance(node, ast.Num):  # Nombre
            return node.n
        elif isinstance(node, ast.BinOp):  # Opération binaire
            return operators[type(node.op)](eval_expr(node.left), eval_expr(node.right))
        elif isinstance(node, ast.UnaryOp):  # Opération unaire
            return operators[type(node.op)](eval_expr(node.operand))
        else:
            raise ValueError(f"Expression non autorisée: {node}")

    try:
        node = ast.parse(expression, mode='eval')
        return str(eval_expr(node.body))
    except Exception as e:
        return f"Erreur de calcul: {str(e)}"

tools = [
    Tool(
        name="Search",
        func=search_web,
        description="Utile pour rechercher des informations actuelles sur le web. Input: une requête de recherche."
    ),
    Tool(
        name="Calculator",
        func=calculate,
        description="Utile pour effectuer des calculs mathématiques. Input: une expression Python (ex: '2+2' ou 'sqrt(16)')."
    )
]

# 3. Prompt ReAct standard
prompt = PromptTemplate.from_template("""
Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Question: {input}
{agent_scratchpad}
""")

# 4. Création de l'agent
agent = create_react_agent(llm, tools, prompt)

# 5. Executor (gère l'exécution)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # Affiche les traces
    max_iterations=5,  # Limite de sécurité
    handle_parsing_errors=True  # Gestion des erreurs
)

# 6. Utilisation
result = agent_executor.invoke({
    "input": "Quelle est la racine carrée de 144 et qui a inventé les mathématiques ?"
})

print(result["output"])

> Entering new AgentExecutor chain...
Thought: I need to calculate the square root of 144 and find who invented mathematics
Action: Calculator
Action Input: sqrt(144)
Observation: 12.0
Thought: Now I need to find who invented mathematics
Action: Search
Action Input: who invented mathematics
Observation: Mathematics wasn't "invented" by one person, but developed over millennia...
Thought: I now have all the information needed
Final Answer: La racine carrée de 144 est 12. Les mathématiques n'ont pas été inventées par une seule personne, mais se sont développées au fil des millénaires...

from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

# Recherche web avec DuckDuckGo
search = DuckDuckGoSearchRun()

# Wikipedia
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

tools = [
    Tool(
        name="WebSearch",
        func=search.run,
        description="Recherche sur le web pour des informations actuelles"
    ),
    Tool(
        name="Wikipedia",
        func=wikipedia.run,
        description="Recherche sur Wikipedia pour des informations encyclopédiques"
    )
]

from langchain_experimental.tools import PythonREPLTool
from langchain.tools import ShellTool

# Exécution de code Python
python_repl = PythonREPLTool()

# Exécution de commandes shell
shell = ShellTool()

tools = [
    Tool(
        name="Python",
        func=python_repl.run,
        description="Exécute du code Python. Utile pour calculs, data science, graphiques."
    ),
    Tool(
        name="Shell",
        func=shell.run,
        description="Exécute des commandes système. ATTENTION: dangereux en production."
    )
]

from langchain.agents import create_sql_agent
from langchain.sql_database import SQLDatabase

# Connexion à une base de données
db = SQLDatabase.from_uri("sqlite:///mydatabase.db")

# Agent SQL automatique
sql_agent = create_sql_agent(
    llm=llm,
    db=db,
    agent_type="openai-tools",
    verbose=True
)

# L'agent génère automatiquement les requêtes SQL
result = sql_agent.run("Combien d'utilisateurs ont plus de 30 ans ?")

import requests

def get_weather(city: str) -> str:
    """Récupère la météo d'une ville"""
    api_key = "YOUR_API_KEY"
    url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}"
    response = requests.get(url)
    data = response.json()
    temp = data["main"]["temp"] - 273.15  # Kelvin vers Celsius
    description = data["weather"][0]["description"]
    return f"À {city}: {temp:.1f}°C, {description}"

weather_tool = Tool(
    name="Weather",
    func=get_weather,
    description="Donne la météo actuelle d'une ville. Input: nom de la ville en anglais."
)

from langchain.tools.retriever import create_retriever_tool

# Supposons que vous avez un retriever RAG configuré
retriever = vectorstore.as_retriever()

# Transformer le retriever en outil d'agent
retriever_tool = create_retriever_tool(
    retriever,
    name="DocumentSearch",
    description="Recherche dans la documentation interne de l'entreprise. Utile pour les questions sur les procédures, politiques, et produits."
)

tools = [search, calculator, retriever_tool]

from langchain.agents import create_react_agent

agent = create_react_agent(llm, tools, prompt)

from langchain.agents import create_openai_functions_agent

agent = create_openai_functions_agent(llm, tools, prompt)

from langchain.agents import create_structured_chat_agent

agent = create_structured_chat_agent(llm, tools, prompt)

from langchain.agents import create_self_ask_with_search_agent

agent = create_self_ask_with_search_agent(llm, [search_tool], prompt)

from langchain.agents import PlanAndExecute
from langchain_experimental.plan_and_execute import load_agent_executor

agent_executor = load_agent_executor(llm, tools, verbose=True)

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_experimental.tools import PythonREPLTool
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

# Outils
search = DuckDuckGoSearchRun()
python = PythonREPLTool()

tools = [
    Tool(name="Search", func=search.run, description="Recherche web"),
    Tool(name="Python", func=python.run, description="Exécute du Python pour calculs et visualisations")
]

# Prompt spécialisé
prompt = ChatPromptTemplate.from_messages([
    ("system", """Tu es un analyste de données expert. Tu peux:
    1. Rechercher des informations sur le web
    2. Effectuer des calculs et créer des graphiques avec Python

    Réponds de manière structurée avec des chiffres précis."""),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad")
])

# LLM configuré une seule fois
llm = ChatOpenAI(model="gpt-4", temperature=0)

# Agent
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Utilisation
result = agent_executor.invoke({
    "input": """
    Recherche les revenus 2024 d'Apple, Google et Microsoft.
    Calcule la moyenne et crée un graphique en barres comparatif.
    Sauvegarde le graphique dans 'comparison.png'.
    """
})

from langchain.tools.retriever import create_retriever_tool
from langchain.agents import create_openai_functions_agent
from langchain.memory import ConversationBufferMemory

# Outil RAG sur la documentation produit
doc_retriever = vectorstore.as_retriever()
doc_tool = create_retriever_tool(
    doc_retriever,
    "ProductDocs",
    "Recherche dans la documentation produit et les FAQs"
)

# Outil de ticket
def create_ticket(issue: str) -> str:
    """Crée un ticket de support"""
    ticket_id = generate_ticket_id()
    save_to_database(ticket_id, issue)
    return f"Ticket #{ticket_id} créé avec succès"

ticket_tool = Tool(
    name="CreateTicket",
    func=create_ticket,
    description="Crée un ticket si le problème ne peut pas être résolu immédiatement"
)

tools = [doc_tool, ticket_tool]

# Mémoire conversationnelle
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# Agent avec mémoire
prompt = ChatPromptTemplate.from_messages([
    ("system", """Tu es un assistant support client.
    1. Cherche d'abord dans la documentation
    2. Si tu ne trouves pas de solution, crée un ticket
    3. Sois courtois et professionnel"""),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad")
])

agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

# Conversation
agent_executor.invoke({"input": "Comment réinitialiser mon mot de passe ?"})
# → Recherche dans la doc, trouve la procédure, explique

agent_executor.invoke({"input": "Ça ne fonctionne toujours pas"})
# → Utilise le contexte précédent, crée un ticket

from langchain.tools import Tool
from datetime import datetime

def fetch_rss_feeds(topics: str) -> str:
    """Récupère les derniers articles RSS"""
    # Implémentation avec feedparser
    return "Articles récents sur ..."

def summarize_article(url: str) -> str:
    """Résume un article web"""
    # Implémentation avec BeautifulSoup + LLM
    return "Résumé: ..."

def save_to_notion(content: str) -> str:
    """Sauvegarde dans Notion"""
    # Implémentation avec l'API Notion
    return "Sauvegardé dans Notion"

tools = [
    Tool(name="FetchNews", func=fetch_rss_feeds, description="Récupère les dernières actualités"),
    Tool(name="Summarize", func=summarize_article, description="Résume un article"),
    Tool(name="SaveNotion", func=save_to_notion, description="Sauvegarde dans Notion")
]

# Agent de veille
prompt = ChatPromptTemplate.from_messages([
    ("system", """Tu es un agent de veille technologique.
    Tâches quotidiennes:
    1. Chercher les actualités sur {topics}
    2. Résumer les 5 articles les plus importants
    3. Sauvegarder dans Notion avec tags appropriés"""),
    ("human", "Effectue la veille du jour"),
    MessagesPlaceholder("agent_scratchpad")
])

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Exécution automatique (via cron ou scheduler)
result = agent_executor.invoke({
    "input": f"Effectue la veille pour {datetime.now().strftime('%Y-%m-%d')}"
})

from langchain.agents import AgentExecutor

# Agent 1 : Recherche
research_agent = create_react_agent(llm, [search_tool], research_prompt)
research_executor = AgentExecutor(agent=research_agent, tools=[search_tool])

# Agent 2 : Analyse
analysis_agent = create_react_agent(llm, [python_tool], analysis_prompt)
analysis_executor = AgentExecutor(agent=analysis_agent, tools=[python_tool])

# Agent 3 : Rédaction
writing_agent = create_react_agent(llm, [], writing_prompt)
writing_executor = AgentExecutor(agent=writing_agent, tools=[])

# Orchestrateur (meta-agent)
def delegate_to_research(query: str) -> str:
    return research_executor.invoke({"input": query})["output"]

def delegate_to_analysis(data: str) -> str:
    return analysis_executor.invoke({"input": data})["output"]

def delegate_to_writing(outline: str) -> str:
    return writing_executor.invoke({"input": outline})["output"]

meta_tools = [
    Tool(name="Research", func=delegate_to_research, description="Agent de recherche"),
    Tool(name="Analysis", func=delegate_to_analysis, description="Agent d'analyse"),
    Tool(name="Writing", func=delegate_to_writing, description="Agent de rédaction")
]

# Meta-agent qui coordonne
meta_agent = create_react_agent(llm, meta_tools, meta_prompt)
meta_executor = AgentExecutor(agent=meta_agent, tools=meta_tools, verbose=True)

# Utilisation
result = meta_executor.invoke({
    "input": "Crée un rapport complet sur les tendances IA 2025"
})
# → Délègue la recherche, puis l'analyse, puis la rédaction

from langchain.agents import AgentExecutor

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=10,
    max_execution_time=60,  # Timeout après 60s
    early_stopping_method="generate",  # Génère une réponse même si incomplet
    handle_parsing_errors=True,  # Gère les erreurs de parsing
    return_intermediate_steps=True  # Retourne tous les steps pour debugging
)

# Avec retry manuel
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def run_agent_with_retry(query):
    return agent_executor.invoke({"input": query})

result = run_agent_with_retry("Question complexe")

from pydantic import BaseModel, Field, validator

class WeatherOutput(BaseModel):
    """Modèle validé pour la météo"""
    temperature: float = Field(..., ge=-50, le=60, description="Température en °C")
    description: str = Field(..., min_length=3)

    @validator('temperature')
    def temp_must_be_reasonable(cls, v):
        if not -50 <= v <= 60:
            raise ValueError('Température hors limites')
        return v

def safe_weather_tool(city: str) -> str:
    """Outil météo avec validation"""
    raw_data = get_weather_api(city)
    validated = WeatherOutput(**raw_data)  # Pydantic valide
    return f"{validated.temperature}°C, {validated.description}"

from functools import lru_cache
from datetime import datetime, timedelta

# Cache simple avec TTL
cache = {}
cache_ttl = {}

def cached_tool(func):
    def wrapper(query: str) -> str:
        # Vérifie le cache
        if query in cache:
            if datetime.now() - cache_ttl[query] < timedelta(hours=1):
                print(f"[CACHE HIT] {query}")
                return cache[query]

        # Exécute et cache
        result = func(query)
        cache[query] = result
        cache_ttl[query] = datetime.now()
        return result

    return wrapper

@cached_tool
def expensive_search(query: str) -> str:
    """Recherche coûteuse avec cache"""
    return search_api(query)

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# LLM avec streaming
llm = ChatOpenAI(
    model="gpt-4",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

# Agent avec streaming
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# Les réponses s'affichent token par token
result = agent_executor.invoke({"input": "Question longue..."})

from ratelimit import limits, sleep_and_retry
import time

# Limite : 10 appels par minute
@sleep_and_retry
@limits(calls=10, period=60)
def rate_limited_tool(query: str) -> str:
    """Outil avec rate limiting"""
    return expensive_api_call(query)

# Compteur de tokens/coût
class CostTracker:
    def __init__(self, budget_usd=10):
        self.total_tokens = 0
        self.total_cost = 0
        self.budget = budget_usd

    def track(self, tokens: int, cost_per_1k: float = 0.03):
        self.total_tokens += tokens
        self.total_cost += (tokens / 1000) * cost_per_1k

        if self.total_cost > self.budget:
            raise Exception(f"Budget dépassé : ${self.total_cost:.2f}")

        print(f"Coût actuel : ${self.total_cost:.2f}/{self.budget}")

tracker = CostTracker(budget_usd=5)

# Callback personnalisé
class CostTrackingCallback:
    def on_llm_end(self, response, **kwargs):
        tokens = response.llm_output.get('token_usage', {}).get('total_tokens', 0)
        tracker.track(tokens)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[CostTrackingCallback()]
)

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("langchain.agents")

# Ou avec LangSmith (recommandé)
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "mon-projet-agents"

from langchain.callbacks.base import BaseCallbackHandler

class DebugCallback(BaseCallbackHandler):
    def on_agent_action(self, action, **kwargs):
        print(f"\n🤖 Action: {action.tool}")
        print(f"📝 Input: {action.tool_input}")

    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f"\n🔧 Outil lancé: {serialized.get('name', 'unknown')}")

    def on_tool_end(self, output, **kwargs):
        print(f"✅ Résultat: {output[:100]}...")

    def on_tool_error(self, error, **kwargs):
        print(f"❌ Erreur: {error}")

    def on_agent_finish(self, finish, **kwargs):
        print(f"\n🏁 Agent terminé")
        print(f"📊 Iterations: {finish.return_values}")

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[DebugCallback()],
    verbose=True
)

result = agent_executor.invoke(
    {"input": "Question"},
    return_intermediate_steps=True
)

# Afficher toutes les étapes
for i, (action, observation) in enumerate(result["intermediate_steps"]):
    print(f"\n--- Étape {i+1} ---")
    print(f"Outil: {action.tool}")
    print(f"Input: {action.tool_input}")
    print(f"Output: {observation[:200]}")

# Compter les appels par outil
from collections import Counter
tool_usage = Counter([action.tool for action, _ in result["intermediate_steps"]])
print(f"\nUtilisation des outils: {tool_usage}")

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=10,  # Force l'arrêt après 10 itérations
    max_execution_time=60  # Timeout après 60 secondes
)

Tool(
    name="Weather",  # Nom court et clair
    func=get_weather,
    description="""Donne la météo ACTUELLE d'une ville.
    Input: Nom de la ville en anglais (ex: 'Paris', 'London').
    Ne PAS utiliser pour des prévisions ou des données historiques."""
)

# Prompt strict
prompt = PromptTemplate.from_template("""
Tu dois UNIQUEMENT utiliser les outils listés ci-dessous.
N'invente JAMAIS un outil qui n'est pas dans cette liste.

Outils disponibles:
{tools}
...
""")

# 1. Utiliser un modèle moins cher pour le raisonnement
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)  # 60x moins cher que GPT-4

# 2. Limiter les itérations
agent_executor = AgentExecutor(agent=agent, tools=tools, max_iterations=5)

# 3. Caching agressif
from langchain.cache import SQLiteCache
from langchain.globals import set_llm_cache
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

# 4. Budget hard limit
class BudgetExceededError(Exception):
    pass

class BudgetCallback(BaseCallbackHandler):
    def __init__(self, max_usd=5):
        self.total_cost = 0
        self.max_cost = max_usd

    def on_llm_end(self, response, **kwargs):
        tokens = response.llm_output.get('token_usage', {}).get('total_tokens', 0)
        self.total_cost += (tokens / 1000) * 0.03  # $0.03 per 1K tokens

        if self.total_cost > self.max_cost:
            raise BudgetExceededError(f"Budget dépassé: ${self.total_cost:.2f}")

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    handle_parsing_errors=True,  # Gère automatiquement
    # Ou message personnalisé:
    handle_parsing_errors="Vérifie ton format de réponse et réessaie."
)

# 1. Environnement sandboxé
from langchain_experimental.tools import PythonREPLTool

python = PythonREPLTool()
python.globals = {"__builtins__": {}}  # Désactive les built-ins dangereux

# 2. Validation des inputs
def safe_shell(command: str) -> str:
    """Shell avec whitelist"""
    allowed_commands = ["ls", "pwd", "date", "whoami"]
    cmd = command.split()[0]

    if cmd not in allowed_commands:
        return f"Commande '{cmd}' non autorisée"

    # Exécute seulement si autorisé
    return subprocess.check_output(command, shell=True).decode()

# 3. Audit logging
class SecurityCallback(BaseCallbackHandler):
    def on_tool_start(self, serialized, input_str, **kwargs):
        tool_name = serialized.get("name")
        if tool_name in ["Shell", "Python"]:
            log_security_event(f"Execution: {tool_name} - {input_str}")