Architecture des Agents IA : Agents Autonomes et Framework LangChain

Les Agents IA sont des systèmes d’intelligence artificielle capables de prendre des décisions de manière autonome, d'utiliser des outils et de résoudre des tâches complexes étape par étape. Dans ce guide, nous examinons les architectures d'agents et leurs implémentations.

Qu’est-ce qu’un Agent ?

Une comparaison simple entre un chatbot et un agent :

1Chatbot:
2User: "How is the weather today?"
3Bot: "I don't know the weather." (static response)
4
5Agent:
6User: "How is the weather today?"
7Agent: 
8  1. "I need to find out the weather."
9  2. Call weather_api.get_weather()
10  3. Interpret the result
11  4. "It's 15°C in Istanbul, partly cloudy."

Composants d’un Agent

1. LLM (Cerveau)

Prise de décision et raisonnement.

2. Outils

Interaction avec le monde extérieur :

Recherche web
Appels API
Requêtes base de données
Exécution de code

3. Mémoire

Historique de conversation et stockage d’informations.

4. Planification

Décomposition des tâches et stratégie.

Modèle ReAct

Cycle Raisonnement + Action :

1Thought: What do I need to do?
2Action: Which tool should I use?
3Action Input: Tool parameters
4Observation: Tool result
5... (repeat)
6Final Answer: Final response

Exemple ReAct

1Question: "What is Tesla's current stock price and how much has it changed in the last month?"
2
3Thought: I need to find Tesla's stock price.
4Action: stock_price
5Action Input: {"symbol": "TSLA"}
6Observation: Current price: $248.50
7
8Thought: Now I need to calculate the 1-month change.
9Action: stock_history  
10Action Input: {"symbol": "TSLA", "period": "1mo"}
11Observation: 1 month ago: $235.20, Change: +5.65%
12
13Thought: I can now answer the question.
14Final Answer: Tesla's current stock price is $248.50. 
15It has shown a 5.65% increase in the last month.

Développement d’Agents avec LangChain

Agent Basique

1from langchain.agents import create_react_agent, AgentExecutor
2from langchain.tools import Tool
3from langchain_openai import ChatOpenAI
4from langchain import hub
5
6# LLM
7llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
8
9# Tools
10def search_web(query: str) -> str:
11    """Searches the web"""
12    # Web search implementation
13    return f"Search results for: {query}"
14
15def calculate(expression: str) -> str:
16    """Performs mathematical calculation"""
17    return str(eval(expression))
18
19tools = [
20    Tool(name="search", func=search_web, description="Searches the web"),
21    Tool(name="calculator", func=calculate, description="Performs math calculations"),
22]
23
24# Prompt
25prompt = hub.pull("hwchase17/react")
26
27# Agent
28agent = create_react_agent(llm, tools, prompt)
29agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
30
31# Run
32result = agent_executor.invoke({"input": "What is the world population in 2024?"})

Outils Personnalisés

1from langchain.tools import BaseTool
2from pydantic import BaseModel, Field
3
4class WeatherInput(BaseModel):
5    city: str = Field(description="City name")
6    unit: str = Field(default="celsius", description="Temperature unit")
7
8class WeatherTool(BaseTool):
9    name = "weather"
10    description = "Fetches weather information"
11    args_schema = WeatherInput
12    
13    def _run(self, city: str, unit: str = "celsius") -> str:
14        # API call
15        weather_data = fetch_weather(city)
16        return f"{city}: {weather_data['temp']}°{'C' if unit == 'celsius' else 'F'}"
17    
18    async def _arun(self, city: str, unit: str = "celsius") -> str:
19        return await async_fetch_weather(city)
20## Systèmes de mémoire
21
22### Mémoire Conversation Buffer
23
24```python
25from langchain.memory import ConversationBufferMemory
26
27memory = ConversationBufferMemory(
28    memory_key="chat_history",
29    return_messages=True
30)
31
32# Add to Agent
33agent_executor = AgentExecutor(
34    agent=agent,
35    tools=tools,
36    memory=memory,
37    verbose=True
38)

Mémoire de résumé

Résumé pour les longues conversations :

1from langchain.memory import ConversationSummaryMemory
2
3memory = ConversationSummaryMemory(
4    llm=llm,
5    memory_key="chat_history"
6)

Mémoire avec Vector Store

Mémoire à long terme avec recherche sémantique :

1from langchain.memory import VectorStoreRetrieverMemory
2from langchain.vectorstores import Chroma
3
4vectorstore = Chroma(embedding_function=embeddings)
5retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
6
7memory = VectorStoreRetrieverMemory(retriever=retriever)

Systèmes multi-agents

Modèle Superviseur

1from langgraph.graph import StateGraph, END
2
3def supervisor_agent(state):
4    """Supervisor distributing tasks"""
5    task = state["task"]
6    
7    if "research" in task:
8        return "researcher"
9    elif "write" in task:
10        return "writer"
11    else:
12        return "general"
13
14def researcher_agent(state):
15    """Agent conducting research"""
16    # Research logic
17    return {"research_results": results}
18
19def writer_agent(state):
20    """Agent writing content"""
21    # Writing logic
22    return {"content": content}
23
24# Create Graph
25workflow = StateGraph(AgentState)
26workflow.add_node("supervisor", supervisor_agent)
27workflow.add_node("researcher", researcher_agent)
28workflow.add_node("writer", writer_agent)
29
30workflow.add_conditional_edges("supervisor", route_to_agent)
31workflow.add_edge("researcher", "supervisor")
32workflow.add_edge("writer", END)
33
34app = workflow.compile()

Agents collaboratifs

1┌─────────────────────────────────────────────────────┐
2│                    Orchestrateur                     │
3└───────────────────────┬─────────────────────────────┘
4                        │
5        ┌───────────────┼───────────────┐
6        │               │               │
7┌───────▼───────┐ ┌─────▼─────┐ ┌──────▼──────┐
8│  Chercheur    │ │  Analyste │ │   Rédacteur  │
9│    Agent      │ │   Agent   │ │    Agent     │
10└───────────────┘ └───────────┘ └─────────────┘

Stratégies de planification

Plan-and-Execute

1from langchain_experimental.plan_and_execute import (
2    PlanAndExecute,
3    load_agent_executor,
4    load_chat_planner
5)
6
7planner = load_chat_planner(llm)
8executor = load_agent_executor(llm, tools, verbose=True)
9
10agent = PlanAndExecute(planner=planner, executor=executor)
11
12result = agent.run("Research and create a draft for a blog post")

Décomposition de tâches

1Tâche principale : "Analyser les produits pour un site e-commerce"
2
3Sous-tâches :
41. Rechercher les produits des concurrents
52. Effectuer une comparaison des prix
63. Analyser les avis clients
74. Générer un rapport
8
9Chaque sous-tâche est gérée par un agent.
10## Gestion des erreurs et reprise
11
12```python
13from langchain.callbacks import get_openai_callback
14
15class RobustAgentExecutor:
16    def __init__(self, agent_executor, max_retries=3):
17        self.executor = agent_executor
18        self.max_retries = max_retries
19    
20    def invoke(self, input_dict):
21        for attempt in range(self.max_retries):
22            try:
23                with get_openai_callback() as cb:
24                    result = self.executor.invoke(input_dict)
25                    
26                    # Token limit check
27                    if cb.total_tokens > 10000:
28                        self.summarize_memory()
29                    
30                    return result
31                    
32            except Exception as e:
33                if attempt == self.max_retries - 1:
34                    raise
35                
36                # Strategy based on Error
37                if "rate_limit" in str(e):
38                    time.sleep(60)
39                elif "context_length" in str(e):
40                    self.truncate_context()
41                else:
42                    time.sleep(2 ** attempt)

Streaming et temps réel

1from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
2
3llm = ChatOpenAI(
4    streaming=True,
5    callbacks=[StreamingStdOutCallbackHandler()]
6)
7
8async def stream_agent_response(query):
9    async for event in agent_executor.astream_events(
10        {"input": query},
11        version="v1"
12    ):
13        kind = event["event"]
14        if kind == "on_chat_model_stream":
15            print(event["data"]["chunk"].content, end="")
16        elif kind == "on_tool_start":
17            print(f"\n🔧 Using tool: {event['name']}")

Considérations pour la production

Journalisation et monitoring

1import logging
2
3class AgentLogger:
4    def __init__(self):
5        self.logger = logging.getLogger("agent")
6    
7    def log_action(self, action, input_data, output):
8        self.logger.info({
9            "action": action,
10            "input": input_data,
11            "output_length": len(str(output)),
12            "timestamp": datetime.now().isoformat()
13        })
14    
15    def log_error(self, error, context):
16        self.logger.error({
17            "error": str(error),
18            "context": context
19        })

Gestion des coûts

1def estimate_cost(agent_run):
2    # Cost per token (GPT)
3    input_cost = agent_run.input_tokens * 0.00003
4    output_cost = agent_run.output_tokens * 0.00006
5    
6    return input_cost + output_cost
7
8# Budget limit
9MAX_BUDGET_PER_REQUEST = 0.50  # $0.50
10
11if estimate_cost(run) > MAX_BUDGET_PER_REQUEST:
12    raise BudgetExceededError()

Conclusion

Les AI Agents permettent des applications d’intelligence artificielle autonomes et proactives. Des frameworks comme LangChain et LangGraph facilitent le développement de systèmes d’agents prêts pour la production.

Chez Veni AI, nous développons des solutions d’agents pour les entreprises.

Architecture des agents IA : agents autonomes et framework LangChain