LangChain Agents en Pratique

pip install langchain langchain-openai langchain-community
pip install wikipedia tavily-python  # Outils optionnels

from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain import hub
import wikipedia

# Définir les outils
def search_wikipedia_tool(query: str) -> str:
    """Recherche sur Wikipédia."""
    try:
        return wikipedia.summary(query, sentences=3)
    except:
        return "Aucun résultat trouvé"

def calculator_tool(expression: str) -> str:
    """Calcule une expression mathématique."""
    try:
        return str(eval(expression))
    except:
        return "Erreur de calcul"

tools = [
    Tool(
        name="Wikipedia",
        func=search_wikipedia_tool,
        description="Utile pour rechercher des informations factuelles sur Wikipédia. Input: une requête de recherche."
    ),
    Tool(
        name="Calculator",
        func=calculator_tool,
        description="Utile pour effectuer des calculs mathématiques. Input: une expression mathématique comme '2+2' ou '15*7'."
    )
]

# Créer le LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)

# Récupérer le prompt ReAct depuis le hub
prompt = hub.pull("hwchase17/react")

# Créer l'agent
agent = create_react_agent(llm, tools, prompt)

# Créer l'executor (qui gère la boucle d'exécution)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # Affiche les étapes
    max_iterations=10,
    handle_parsing_errors=True
)

# Exécuter
result = agent_executor.invoke({
    "input": "Qui a inventé le téléphone ? En quelle année ? Combien d'années se sont écoulées depuis ?"
})

print(f"\n🎯 Résultat final : {result['output']}")

> Entering new AgentExecutor chain...

I need to find out who invented the telephone and when.

Action: Wikipedia
Action Input: "invention téléphone"

Observation: Le téléphone a été inventé par Alexander Graham Bell en 1876...

Thought: Now I need to calculate how many years have passed since 1876.

Action: Calculator
Action Input: 2024 - 1876

Observation: 148

Thought: I now know the final answer.

Final Answer: Alexander Graham Bell a inventé le téléphone en 1876. 148 années se sont écoulées depuis.

> Finished chain.

🎯 Résultat final : Alexander Graham Bell a inventé le téléphone en 1876. 148 années se sont écoulées depuis.

from langchain.agents import create_openai_functions_agent

# Même configuration des outils et LLM

# Récupérer le prompt OpenAI Functions
prompt = hub.pull("hwchase17/openai-functions-agent")

# Créer l'agent
agent = create_openai_functions_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True
)

result = agent_executor.invoke({"input": "Quelle est la population du Japon ? Divisée par 2 ?"})

from langchain.agents import create_structured_chat_agent
from langchain.memory import ConversationBufferMemory

# Mémoire pour garder l'historique
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

prompt = hub.pull("hwchase17/structured-chat-agent")

agent = create_structured_chat_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

# Conversation multi-tours
agent_executor.invoke({"input": "Cherche la population de Paris"})
agent_executor.invoke({"input": "Et celle de Lyon ?"})  # Comprend le contexte
agent_executor.invoke({"input": "Quelle est la différence ?"})  # Utilise les deux résultats précédents

from langchain.agents import create_self_ask_with_search_agent
from langchain.tools import TavilySearchResults

# Outil de recherche (Tavily est une API de recherche web)
search = TavilySearchResults(max_results=3)

tools = [search]

prompt = hub.pull("hwchase17/self-ask-with-search")

agent = create_self_ask_with_search_agent(llm, tools, prompt)

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({
    "input": "Quel âge a le président actuel de la France ?"
})

Question: Quel âge a le président actuel de la France ?
Are follow up questions needed: Yes
Follow up: Qui est le président actuel de la France ?
Intermediate answer: Emmanuel Macron
Follow up: Quel âge a Emmanuel Macron ?
Intermediate answer: 46 ans (né en 1977)
Final answer: 46 ans

from langchain.agents import PlanAndExecute
from langchain_experimental.plan_and_execute import (
    PlanAndExecute,
    load_agent_executor,
    load_chat_planner
)

# Créer le planificateur
planner = load_chat_planner(llm)

# Créer l'exécuteur
executor = load_agent_executor(llm, tools, verbose=True)

# Créer l'agent Plan-and-Execute
agent = PlanAndExecute(planner=planner, executor=executor, verbose=True)

result = agent.run("""
Analyse le marché des voitures électriques :
1. Les 3 principaux constructeurs
2. Parts de marché
3. Évolution des ventes en 2024
""")

from langchain.agents import AgentType
from langchain_community.agent_toolkits import create_openai_assistant_agent

# Créer l'assistant (côté OpenAI)
assistant = openai.beta.assistants.create(
    name="Data Analyst",
    instructions="Tu es un analyste de données expert.",
    tools=[{"type": "code_interpreter"}],
    model="gpt-4"
)

# Créer l'agent LangChain
agent = create_openai_assistant_agent(
    assistant_id=assistant.id,
    tools=tools
)

# Utilisation
result = agent.invoke({"input": "Analyse ce CSV..."})

from langchain.tools import (
    WikipediaQueryRun,
    DuckDuckGoSearchRun,
    YouTubeSearchTool,
    ArxivQueryRun,
    ShellTool,
    PythonREPLTool
)
from langchain_community.tools import (
    GoogleSerperAPIWrapper,
    TavilySearchResults,
    BraveSearch
)
from langchain_community.utilities import (
    SQLDatabase,
    SerpAPIWrapper
)

# Exemple : Recherche web avec Tavily
search = TavilySearchResults(
    max_results=5,
    search_depth="advanced",  # "basic" ou "advanced"
    include_answer=True,
    include_raw_content=False
)

# Exemple : Requêtes SQL
db = SQLDatabase.from_uri("sqlite:///my_database.db")

from langchain_community.tools.sql_database.tool import (
    QuerySQLDataBaseTool,
    InfoSQLDatabaseTool,
    ListSQLDatabaseTool
)

sql_tools = [
    QuerySQLDataBaseTool(db=db),
    InfoSQLDatabaseTool(db=db),
    ListSQLDatabaseTool(db=db)
]

# Créer un agent SQL
agent = create_openai_functions_agent(llm, sql_tools, prompt)

from langchain.tools import tool

@tool
def get_exchange_rate(from_currency: str, to_currency: str) -> str:
    """Obtient le taux de change entre deux devises.

    Args:
        from_currency: Code de la devise source (ex: "USD")
        to_currency: Code de la devise cible (ex: "EUR")
    """
    # Appel API réel ici
    return f"1 {from_currency} = 0.92 {to_currency}"

# Utilisation
tools = [get_exchange_rate]

from langchain.tools import BaseTool
from pydantic import Field
from typing import Optional, Type
from pydantic import BaseModel

class ExchangeRateInput(BaseModel):
    """Input pour l'outil ExchangeRate"""
    from_currency: str = Field(description="Code de la devise source (ex: USD)")
    to_currency: str = Field(description="Code de la devise cible (ex: EUR)")

class ExchangeRateTool(BaseTool):
    name = "exchange_rate"
    description = "Obtient le taux de change entre deux devises"
    args_schema: Type[BaseModel] = ExchangeRateInput

    def _run(self, from_currency: str, to_currency: str) -> str:
        """Exécution synchrone"""
        # Logique ici
        return f"1 {from_currency} = 0.92 {to_currency}"

    async def _arun(self, from_currency: str, to_currency: str) -> str:
        """Exécution asynchrone"""
        # Logique async ici
        return await self._run(from_currency, to_currency)

# Utilisation
tool = ExchangeRateTool()

from langchain.tools import StructuredTool
from pydantic import BaseModel, Field

class EmailInput(BaseModel):
    to: str = Field(description="Adresse email du destinataire")
    subject: str = Field(description="Sujet de l'email")
    body: str = Field(description="Corps de l'email")
    cc: Optional[list[str]] = Field(default=None, description="Personnes en copie")

def send_email(to: str, subject: str, body: str, cc: Optional[list] = None) -> str:
    """Envoie un email via SMTP"""
    # Logique d'envoi
    return f"Email envoyé à {to}"

email_tool = StructuredTool.from_function(
    func=send_email,
    name="SendEmail",
    description="Envoie un email. Utilise ce tool uniquement si l'utilisateur demande explicitement d'envoyer un email.",
    args_schema=EmailInput
)

from langchain_community.agent_toolkits import (
    SQLDatabaseToolkit,
    FileManagementToolkit,
    GmailToolkit,
    JiraToolkit,
    SlackToolkit
)

# Exemple : Toolkit SQL
db = SQLDatabase.from_uri("postgresql://user:pass@localhost/db")
toolkit = SQLDatabaseToolkit(db=db, llm=llm)

tools = toolkit.get_tools()
# Retourne : QuerySQLTool, InfoSQLTool, ListTablesTool, etc.

agent = create_openai_functions_agent(llm, tools, prompt)

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory
)

# Conversation
agent_executor.invoke({"input": "Je m'appelle Alice"})
agent_executor.invoke({"input": "Comment je m'appelle ?"})
# → "Tu t'appelles Alice"

from langchain.memory import ConversationBufferWindowMemory

memory = ConversationBufferWindowMemory(
    k=5,  # Garder les 5 derniers échanges
    memory_key="chat_history",
    return_messages=True
)

from langchain.memory import ConversationSummaryMemory

memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="chat_history",
    return_messages=True
)

from langchain.memory import VectorStoreRetrieverMemory
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

# Créer le vector store
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts([], embeddings)

memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
)

# L'agent récupère automatiquement les 3 souvenirs les plus pertinents

from langchain.memory import CombinedMemory

# Court-terme : derniers 5 messages
short_term = ConversationBufferWindowMemory(k=5, memory_key="recent_history")

# Long-terme : recherche vectorielle
long_term = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(),
    memory_key="relevant_history"
)

# Combiner
memory = CombinedMemory(memories=[short_term, long_term])

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # Affiche chaque étape
    return_intermediate_steps=True  # Retourne les étapes dans la réponse
)

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

# Accéder aux étapes intermédiaires
for step in result["intermediate_steps"]:
    action, observation = step
    print(f"Action: {action.tool} - {action.tool_input}")
    print(f"Observation: {observation}\n")

from langchain.callbacks import StdOutCallbackHandler
from langchain.callbacks.base import BaseCallbackHandler

class CustomCallbackHandler(BaseCallbackHandler):
    """Callback personnalisé pour logger"""

    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"🔵 LLM Start: {len(prompts)} prompts")

    def on_llm_end(self, response, **kwargs):
        print(f"🟢 LLM End: {response.generations[0][0].text[:50]}...")

    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f"🔧 Tool Start: {serialized['name']} - {input_str}")

    def on_tool_end(self, output, **kwargs):
        print(f"✅ Tool End: {output[:50]}...")

    def on_agent_action(self, action, **kwargs):
        print(f"🤖 Agent Action: {action.tool}")

# Utilisation
handler = CustomCallbackHandler()

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

# Installation
pip install langsmith

# Configuration
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=your_api_key
export LANGCHAIN_PROJECT=your_project_name

# Le code reste inchangé
agent_executor.invoke({"input": "Question"})

# Automatiquement tracé dans LangSmith

import logging
import json
from datetime import datetime

# Configurer le logger
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class LoggingCallbackHandler(BaseCallbackHandler):
    """Logger structuré pour production"""

    def on_agent_finish(self, finish, **kwargs):
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "type": "agent_finish",
            "output": finish.return_values,
            "log": finish.log
        }
        logger.info(json.dumps(log_entry))

    def on_tool_error(self, error, **kwargs):
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "type": "tool_error",
            "error": str(error)
        }
        logger.error(json.dumps(log_entry))

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

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

# L'agent streame maintenant sa réponse token par token

from langchain.cache import InMemoryCache, SQLiteCache
from langchain.globals import set_llm_cache

# Option 1 : Cache en mémoire
set_llm_cache(InMemoryCache())

# Option 2 : Cache persistant (SQLite)
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

# Les appels identiques sont maintenant cachés

from langchain.agents import AgentExecutor

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_concurrent_operations=3  # Exécute jusqu'à 3 outils en parallèle
)

from langchain.llms.base import BaseLLM
from langchain.callbacks import get_openai_callback
import time

class RateLimitedLLM(BaseLLM):
    """LLM avec rate limiting"""

    def __init__(self, llm, requests_per_minute=20):
        self.llm = llm
        self.requests_per_minute = requests_per_minute
        self.last_request_time = 0

    def _call(self, prompt, stop=None):
        # Attendre si nécessaire
        time_since_last = time.time() - self.last_request_time
        min_interval = 60.0 / self.requests_per_minute

        if time_since_last < min_interval:
            time.sleep(min_interval - time_since_last)

        self.last_request_time = time.time()
        return self.llm._call(prompt, stop)

# Utilisation
llm = RateLimitedLLM(ChatOpenAI(model="gpt-4"), requests_per_minute=20)

from langchain.agents import AgentExecutor

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=10,
    handle_parsing_errors=True,  # Retry sur erreurs de parsing
    early_stopping_method="generate"  # Méthode de fallback
)

from langchain.tools import Tool

def safe_tool_wrapper(tool_func):
    """Wrapper qui gère les erreurs"""
    def wrapper(*args, **kwargs):
        try:
            return tool_func(*args, **kwargs)
        except Exception as e:
            return f"Erreur lors de l'exécution : {str(e)}"
    return wrapper

# Appliquer à tous les outils
safe_tools = [
    Tool(
        name=tool.name,
        func=safe_tool_wrapper(tool.func),
        description=tool.description
    )
    for tool in tools
]

import signal
from contextlib import contextmanager

@contextmanager
def timeout(seconds):
    """Context manager pour timeout"""
    def handler(signum, frame):
        raise TimeoutError(f"Timeout après {seconds}s")

    signal.signal(signal.SIGALRM, handler)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)

# Utilisation
try:
    with timeout(30):  # 30 secondes max
        result = agent_executor.invoke({"input": "Question complexe"})
except TimeoutError:
    result = "Timeout : la requête a pris trop de temps"

from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain.tools import tool
from langchain import hub

# ========== OUTILS ==========

@tool
def check_order_status(order_id: str) -> str:
    """Vérifie le statut d'une commande. Input: order_id (ex: 'CMD-12345')"""
    # Simuler un appel BDD/API
    orders = {
        "CMD-12345": "En cours de livraison",
        "CMD-67890": "Livrée",
        "CMD-11111": "En préparation"
    }
    return orders.get(order_id, "Commande introuvable")

@tool
def initiate_refund(order_id: str, reason: str) -> str:
    """Initie un remboursement. Inputs: order_id, reason"""
    # Logique de remboursement
    return f"Remboursement initié pour {order_id}. Raison: {reason}. Délai: 3-5 jours."

@tool
def search_knowledge_base(query: str) -> str:
    """Recherche dans la base de connaissances. Input: query"""
    kb = {
        "retour": "Politique de retour : 30 jours, article non utilisé",
        "livraison": "Livraison gratuite >50€, 2-5 jours ouvrés",
        "paiement": "Cartes acceptées : Visa, MasterCard, PayPal"
    }
    for key, value in kb.items():
        if key in query.lower():
            return value
    return "Aucune information trouvée"

tools = [check_order_status, initiate_refund, search_knowledge_base]

# ========== AGENT ==========

llm = ChatOpenAI(model="gpt-4", temperature=0)

prompt = hub.pull("hwchase17/openai-functions-agent")

# Personnaliser le prompt système
prompt.messages[0].prompt.template = """Tu es un agent de support client bienveillant et efficace.

Règles :
1. Toujours commencer par identifier le numéro de commande si pertinent
2. Utiliser les outils disponibles pour obtenir des informations précises
3. Si le problème ne peut être résolu, proposer de l'escalader à un humain
4. Être empathique et professionnel

Outils disponibles : {tools}"""

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

agent = create_openai_functions_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True,
    max_iterations=5,
    handle_parsing_errors=True
)

# ========== BOUCLE DE CONVERSATION ==========

print("🤖 Support Client IA - Tapez 'exit' pour quitter\n")

while True:
    user_input = input("👤 Vous : ")

    if user_input.lower() in ['exit', 'quit', 'bye']:
        print("🤖 Agent : Au revoir ! N'hésitez pas à revenir.")
        break

    try:
        response = agent_executor.invoke({"input": user_input})
        print(f"🤖 Agent : {response['output']}\n")
    except Exception as e:
        print(f"❌ Erreur : {str(e)}\n")

🤖 Support Client IA - Tapez 'exit' pour quitter

👤 Vous : Bonjour, ma commande CMD-12345 n'est pas arrivée

> Entering new AgentExecutor chain...

[Agent utilise check_order_status]

🤖 Agent : Bonjour ! J'ai vérifié votre commande CMD-12345. Elle est actuellement en cours de livraison. Vous devriez la recevoir dans les prochains jours. Puis-je vous aider avec autre chose ?

👤 Vous : C'était prévu pour hier, je voudrais un remboursement

[Agent utilise initiate_refund]

🤖 Agent : Je comprends votre frustration. J'ai initié un remboursement pour votre commande CMD-12345 en raison du retard de livraison. Vous recevrez les fonds sous 3 à 5 jours ouvrés. Toutes nos excuses pour ce désagrément.

👤 Vous : Merci !

🤖 Agent : Je vous en prie ! N'hésitez pas à nous contacter si vous avez d'autres questions.