Assistant personnel multi-outils LangChain
Cet article construit un Agent assistant personnel qui intègre la requête météo, la gestion d'agenda et l'envoi d'e-mails, démontrant l'utilisation complète de la collaboration multi-outils et de la sortie structurée.
Conception de systèmes
- Trois outils : requête météo, gestion d'agenda, envoi d'e-mails
- Sortie structurée : résumé du planning formaté en Markdown
- Sortie en streaming : affichage en temps réel de la réflexion et du traitement de l'IA
- Mémoire de conversation : se souvenir des préférences et du contexte de l'utilisateur
Code complet
Avant de lancer, installez les dépendances, et.envconfigurez-le correctement.DEEPSEEK_API_KEY：
pip install langchain langchain-deepseek langgraph-checkpoint-sqlite python-dotenv
Exemple
# Chemin du fichier : personal_assistant.py
from dotenv import load_dotenv
load_dotenv()
import sqlite3
from datetime import datetime
from pydantic import BaseModel, Field
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langgraph.checkpoint.sqlite import SqliteSaver
# ========== 1. Données simulées ==========
calendar_events = [
{"id": 1, "title": "Python Study", "date": "2024-03-25",
"time": "14:00", "duration": "2 hours"},
{"id": 2, "title": "Réunion hebdomadaire d'équipe", "date": "2024-03-25",
"time": "10:00", "duration": "1 heure"},
{"id": 3, "title": "Revue de code", "date": "2024-03-26",
"time": "15:00", "duration": "1.5 heures"},
]
weather_db = {
"Hangzhou": {"condition": "Ensoleillé", "temp": 25, "humidity": 60},
"Beijing": {"condition": "Nuageux", "temp": 18, "humidity": 45},
"Shanghai": {"condition": "Pluie légère", "temp": 22, "humidity": 80},
}
# ========== 2. Définir les outils ==========
@tool
def get_weather(city: str) -> str:
"""Interroger la météo en temps réel d'une ville spécifiée.
Arguments :
city: Nom de la ville, par exemple Hangzhou, Beijing, Shanghai
"""
data = weather_db.get(city)
if not data:
return f"La météo pour {city} n'est pas prise en charge actuellement. Villes prises en charge : {', '.join(weather_db.keys())}"
return (f"Météo de {city} : {data['condition']},"
f"Température {data['temp']}°C, humidité {data['humidity']}%")
@tool
def query_schedule(date: str = None) -> str:
"""Interroger le planning pour une date spécifiée. Si aucune date n'est spécifiée, interroger le planning d'aujourd'hui.
Arguments :
date: Date, format YYYY-MM-DD, par ex., 2024-03-25. Si non fournie, interroger aujourd'hui
"""
if date is None:
date = datetime.now().strftime("%Y-%m-%d")
events = [e for e in calendar_events if e["date"] == date]
if not events:
return f"{date} n'a pas de planning."
events.sort(key=lambda e: e["time"])
# Ici, vous devez écrire directement le caractère emoji, et non une entité HTML comme 📅 —
# Il s'agit d'une chaîne Python ; elle sera imprimée telle quelle et ne sera pas interprétée comme un graphique par le navigateur
lines = [f"📅 {date} Programme :"]
for e in events:
lines.append(f" - {e['time']} {e['title']}（{e['duration']}）")
return "\n".join(lines)
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Envoyer un e-mail (simulé).
Arguments :
to: e-mail du destinataire
subject: Objet de l'e-mail
body: Corps de l'e-mail
"""
# Simuler l'envoi
email_id = f"MSG-{datetime.now().strftime('%Y%m%d%H%M%S')}"
return f"E-mail envoyé ! Destinataire : {to}, Objet : {subject}, ID de l'e-mail : {email_id}"
# ========== 3. Modèle de sortie structuré ==========
class DailySummary(BaseModel):
"""Résumé quotidien"""
date: str = Field(description="Date")
weather_summary: str = Field(description="Aperçu météo")
event_count: int = Field(description="Nombre d'événements")
key_events: list[str] = Field(description="Liste des événements importants")
suggestion: str = Field(description="Suggestion du jour")
def to_markdown(summary: DailySummary) -> str:
"""Rendre le DailySummary structuré sous forme de texte Markdown,
correspondant à l'élément « Sortie structurée : résumé du planning formaté en Markdown » de la conception du système."""
lines = [
f"## {summary.date} Résumé du jour",
"",
f"- **Météo** : {summary.weather_summary}",
f"- **Nombre d'événements** : {summary.event_count}",
"",
"**Éléments importants :**",
]
if summary.key_events:
lines += [f"- {event}" for event in summary.key_events]
else:
lines.append("- None")
lines += ["", f"**Suggestion du jour** : {summary.suggestion}"]
return "\n".join(lines)
# ========== 4. Définir le middleware ==========
@dynamic_prompt
def inject_date_context(request) -> str:
"""Ajoute dynamiquement les informations sur la date actuelle après le prompt système.
L'approche précédente consistant à insérer le message de date dans la liste de messages avec @before_model avait deux problèmes : 72. 1. La mise à jour de message renvoyée par before_model passe par le reducer add_messages, qui ne fait que
1. Les mises à jour de message renvoyées par before_model passent par le reducer add_messages ; le reducer uniquement selon
« ajoute les nouveaux messages à la fin », en ignorant la position dans la liste renvoyée ; insert(-1, ...)
l'intention d'insérer avant le message utilisateur n'a en réalité aucun effet ;
2. before_model est déclenché à plusieurs reprises dans la boucle d'appels d'outils à plusieurs tours, ce qui peut facilement insérer ce message
entre AIMessage(tool_calls=...) et son ToolMessage correspondant,
ce qui perturbe l'exigence stricte d'ordre des messages.
À la place, utilisez @dynamic_prompt pour réécrire directement la chaîne de prompt système ; chaque appel de modèle
la recalculera, sans toucher à la liste de messages, de sorte qu'il n'y a aucun risque d'accumulation ou de désalignement.
"""
now = datetime.now()
weekday = ["Lun", "Mar", "Mer", "Jeu", "Ven", "Sam", "Dim"][now.weekday()]
date_hint = (f"\n\n[Prompt système] La date actuelle est {now.strftime('%Y-%m-%d')},
f"Jour de la semaine {weekday}. Si l'utilisateur ne spécifie pas de date, interrogez la date du jour par défaut.")
return request.system_prompt + date_hint
# ========== 5. Créer un agent ==========
# Utilisez SqliteSaver pour persister les conversations, en implémentant la « mémoire de conversation » dans la conception du système.
# Établissez vous-même la connexion et passez-la au constructeur SqliteSaver, plutôt que d'utiliser
# SqliteSaver.from_conn_string() (qui est un gestionnaire de contexte adapté uniquement aux instructions with,
# un gestionnaire de contexte qui se ferme après utilisation ; voir l'article « Robot de service client intelligent LangChain »).
conn = sqlite3.connect("personal_assistant.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0.3)
agent = create_agent(
model=model,
tools=[get_weather, query_schedule, send_email],
middleware=[inject_date_context],
response=DailySummary,
checkpointer=checkpointer,
system_prompt="""Vous êtes un assistant personnel nommé "Xiao Zhu". Vous pouvez vérifier la météo, gérer les plannings et envoyer des e-mails.
## Comment ça fonctionne
1. Lorsque l’utilisateur demande « Comment se passe aujourd’hui ? » ou des questions similaires :
- D’abord interroger la météo d’aujourd’hui (get_weather)
- Ensuite interroger le planning d’aujourd’hui (query_schedule)
- Puis générer un résumé quotidien
2. Lorsque l’utilisateur demande d’envoyer un e-mail, utilisez l’outil send_email
3. Lorsque l’utilisateur demande uniquement la météo ou uniquement le planning, appelez uniquement l’outil correspondant
## Style
- Le ton est amical et naturel
- Préférez utiliser des outils pour obtenir des données en temps réel ; n'inventez rien""",
)
# ========== 6. Fonction d'interaction ==========
def chat(message: str, thread_id: str = "xiaoming"):
"""Discuter avec l'assistant :
- Utilisez stream_mode="values" pour diffuser la réflexion et le traitement de l'IA (correspondant à « sortie en flux »)
- Transmettez thread_id ; les conversations sous le même thread_id seront mémorisées par SqliteSaver (correspondant à « mémoire de conversation »)
- Si un résumé structuré est généré dans ce tour, il sera rendu en Markdown et imprimé (correspondant à « sortie structurée »)
"""
config = {"configurable": {"thread_id": thread_id}}
print(f"\n{'='*60}")
print(f"Vous : {message}")
print(f"{'='*60}")
seen = 0
final_state = {}
for state in agent.stream(
{"messages": [HumanMessage(content=message)]},
config=config,
stream_mode="values",
):
final_state = state
msgs = state.get("messages", [])
# Chaque fois que quelques nouveaux messages de plus apparaissent, cela signifie que l'Agent a avancé d'une étape ; affichez-les en temps réel
for msg in msgs[seen:]:
if msg.type == "ai" and getattr(msg, "tool_calls", None):
for call in msg.tool_calls:
print(f"🤔 A décidé d'appeler l'outil : {call['name']}, arguments : {call['args']}")
elif msg.type == "tool":
tool_name = getattr(msg, "name", "") or "Outil"
print(f"🔧 {tool_name} a renvoyé : {str(msg.content)[:80]}")
elif msg.type == "ai" and msg.content:
print(f"🤖 Assistant : {msg.content}")
seen = len(msgs)
# Remarque : response_format est configuré au niveau de l'Agent ; chaque appel (y compris l'envoi d'e-mails,
# les demandes de suivi sans rapport avec "schedule summary") tenteront de force de générer un DailySummary,
# Il s'agit d'une limitation connue du response_format global. Pour créer véritablement un assistant multi-intentions, une meilleure approche consiste
# de faire de "generate daily summary" un outil séparé, en laissant l'Agent décider s'il faut l'appeler,
# plutôt que d'attacher un schéma de sortie toujours actif à l'ensemble de l'Agent.
if "structured_response" in final_state:
summary = final_state["structured_response"]
print("\n--- Résumé structuré (Markdown) ---)
print(to_markdown(summary))
return final_state
# ========== 7. Tests ==========
if __name__ == "__main__":
chat(Quel temps fait-il à Hangzhou aujourd'hui ? Consulte mon emploi du temps, puis donne-moi un résumé pour aujourd'hui)
chat(Aide-moi à envoyer un e-mail à [email protected], avec pour objet « Résumé quotidien » et un contenu indiquant que l'emploi du temps d'aujourd'hui a été confirmé)
# Troisième tour : aucune nouvelle information, on teste uniquement si l'Agent se souvient de la conversation précédente —
# Comme le même thread_id est transmis, le checkpointer ramènera l'historique complet
chat(Quel était l'objet de l'e-mail que je t'ai demandé d'envoyer tout à l'heure ?)
from dotenv import load_dotenv
load_dotenv()
import sqlite3
from datetime import datetime
from pydantic import BaseModel, Field
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langgraph.checkpoint.sqlite import SqliteSaver
# ========== 1. Données simulées ==========
calendar_events = [
{"id": 1, "title": "Python Study", "date": "2024-03-25",
"time": "14:00", "duration": "2 hours"},
{"id": 2, "title": "Réunion hebdomadaire d'équipe", "date": "2024-03-25",
"time": "10:00", "duration": "1 heure"},
{"id": 3, "title": "Revue de code", "date": "2024-03-26",
"time": "15:00", "duration": "1.5 heures"},
]
weather_db = {
"Hangzhou": {"condition": "Ensoleillé", "temp": 25, "humidity": 60},
"Beijing": {"condition": "Nuageux", "temp": 18, "humidity": 45},
"Shanghai": {"condition": "Pluie légère", "temp": 22, "humidity": 80},
}
# ========== 2. Définir les outils ==========
@tool
def get_weather(city: str) -> str:
"""Interroger la météo en temps réel d'une ville spécifiée.
Arguments :
city: Nom de la ville, par exemple Hangzhou, Beijing, Shanghai
"""
data = weather_db.get(city)
if not data:
return f"La météo pour {city} n'est pas prise en charge actuellement. Villes prises en charge : {', '.join(weather_db.keys())}"
return (f"Météo de {city} : {data['condition']},"
f"Température {data['temp']}°C, humidité {data['humidity']}%")
@tool
def query_schedule(date: str = None) -> str:
"""Interroger le planning pour une date spécifiée. Si aucune date n'est spécifiée, interroger le planning d'aujourd'hui.
Arguments :
date: Date, format YYYY-MM-DD, par ex., 2024-03-25. Si non fournie, interroger aujourd'hui
"""
if date is None:
date = datetime.now().strftime("%Y-%m-%d")
events = [e for e in calendar_events if e["date"] == date]
if not events:
return f"{date} n'a pas de planning."
events.sort(key=lambda e: e["time"])
# Ici, vous devez écrire directement le caractère emoji, et non une entité HTML comme 📅 —
# Il s'agit d'une chaîne Python ; elle sera imprimée telle quelle et ne sera pas interprétée comme un graphique par le navigateur
lines = [f"📅 {date} Programme :"]
for e in events:
lines.append(f" - {e['time']} {e['title']}（{e['duration']}）")
return "\n".join(lines)
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Envoyer un e-mail (simulé).
Arguments :
to: e-mail du destinataire
subject: Objet de l'e-mail
body: Corps de l'e-mail
"""
# Simuler l'envoi
email_id = f"MSG-{datetime.now().strftime('%Y%m%d%H%M%S')}"
return f"E-mail envoyé ! Destinataire : {to}, Objet : {subject}, ID de l'e-mail : {email_id}"
# ========== 3. Modèle de sortie structuré ==========
class DailySummary(BaseModel):
"""Résumé quotidien"""
date: str = Field(description="Date")
weather_summary: str = Field(description="Aperçu météo")
event_count: int = Field(description="Nombre d'événements")
key_events: list[str] = Field(description="Liste des événements importants")
suggestion: str = Field(description="Suggestion du jour")
def to_markdown(summary: DailySummary) -> str:
"""Rendre le DailySummary structuré sous forme de texte Markdown,
correspondant à l'élément « Sortie structurée : résumé du planning formaté en Markdown » de la conception du système."""
lines = [
f"## {summary.date} Résumé du jour",
"",
f"- **Météo** : {summary.weather_summary}",
f"- **Nombre d'événements** : {summary.event_count}",
"",
"**Éléments importants :**",
]
if summary.key_events:
lines += [f"- {event}" for event in summary.key_events]
else:
lines.append("- None")
lines += ["", f"**Suggestion du jour** : {summary.suggestion}"]
return "\n".join(lines)
# ========== 4. Définir le middleware ==========
@dynamic_prompt
def inject_date_context(request) -> str:
"""Ajoute dynamiquement les informations sur la date actuelle après le prompt système.
L'approche précédente consistant à insérer le message de date dans la liste de messages avec @before_model avait deux problèmes : 72. 1. La mise à jour de message renvoyée par before_model passe par le reducer add_messages, qui ne fait que
1. Les mises à jour de message renvoyées par before_model passent par le reducer add_messages ; le reducer uniquement selon
« ajoute les nouveaux messages à la fin », en ignorant la position dans la liste renvoyée ; insert(-1, ...)
l'intention d'insérer avant le message utilisateur n'a en réalité aucun effet ;
2. before_model est déclenché à plusieurs reprises dans la boucle d'appels d'outils à plusieurs tours, ce qui peut facilement insérer ce message
entre AIMessage(tool_calls=...) et son ToolMessage correspondant,
ce qui perturbe l'exigence stricte d'ordre des messages.
À la place, utilisez @dynamic_prompt pour réécrire directement la chaîne de prompt système ; chaque appel de modèle
la recalculera, sans toucher à la liste de messages, de sorte qu'il n'y a aucun risque d'accumulation ou de désalignement.
"""
now = datetime.now()
weekday = ["Lun", "Mar", "Mer", "Jeu", "Ven", "Sam", "Dim"][now.weekday()]
date_hint = (f"\n\n[Prompt système] La date actuelle est {now.strftime('%Y-%m-%d')},
f"Jour de la semaine {weekday}. Si l'utilisateur ne spécifie pas de date, interrogez la date du jour par défaut.")
return request.system_prompt + date_hint
# ========== 5. Créer un agent ==========
# Utilisez SqliteSaver pour persister les conversations, en implémentant la « mémoire de conversation » dans la conception du système.
# Établissez vous-même la connexion et passez-la au constructeur SqliteSaver, plutôt que d'utiliser
# SqliteSaver.from_conn_string() (qui est un gestionnaire de contexte adapté uniquement aux instructions with,
# un gestionnaire de contexte qui se ferme après utilisation ; voir l'article « Robot de service client intelligent LangChain »).
conn = sqlite3.connect("personal_assistant.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0.3)
agent = create_agent(
model=model,
tools=[get_weather, query_schedule, send_email],
middleware=[inject_date_context],
response=DailySummary,
checkpointer=checkpointer,
system_prompt="""Vous êtes un assistant personnel nommé "Xiao Zhu". Vous pouvez vérifier la météo, gérer les plannings et envoyer des e-mails.
## Comment ça fonctionne
1. Lorsque l’utilisateur demande « Comment se passe aujourd’hui ? » ou des questions similaires :
- D’abord interroger la météo d’aujourd’hui (get_weather)
- Ensuite interroger le planning d’aujourd’hui (query_schedule)
- Puis générer un résumé quotidien
2. Lorsque l’utilisateur demande d’envoyer un e-mail, utilisez l’outil send_email
3. Lorsque l’utilisateur demande uniquement la météo ou uniquement le planning, appelez uniquement l’outil correspondant
## Style
- Le ton est amical et naturel
- Préférez utiliser des outils pour obtenir des données en temps réel ; n'inventez rien""",
)
# ========== 6. Fonction d'interaction ==========
def chat(message: str, thread_id: str = "xiaoming"):
"""Discuter avec l'assistant :
- Utilisez stream_mode="values" pour diffuser la réflexion et le traitement de l'IA (correspondant à « sortie en flux »)
- Transmettez thread_id ; les conversations sous le même thread_id seront mémorisées par SqliteSaver (correspondant à « mémoire de conversation »)
- Si un résumé structuré est généré dans ce tour, il sera rendu en Markdown et imprimé (correspondant à « sortie structurée »)
"""
config = {"configurable": {"thread_id": thread_id}}
print(f"\n{'='*60}")
print(f"Vous : {message}")
print(f"{'='*60}")
seen = 0
final_state = {}
for state in agent.stream(
{"messages": [HumanMessage(content=message)]},
config=config,
stream_mode="values",
):
final_state = state
msgs = state.get("messages", [])
# Chaque fois que quelques nouveaux messages de plus apparaissent, cela signifie que l'Agent a avancé d'une étape ; affichez-les en temps réel
for msg in msgs[seen:]:
if msg.type == "ai" and getattr(msg, "tool_calls", None):
for call in msg.tool_calls:
print(f"🤔 A décidé d'appeler l'outil : {call['name']}, arguments : {call['args']}")
elif msg.type == "tool":
tool_name = getattr(msg, "name", "") or "Outil"
print(f"🔧 {tool_name} a renvoyé : {str(msg.content)[:80]}")
elif msg.type == "ai" and msg.content:
print(f"🤖 Assistant : {msg.content}")
seen = len(msgs)
# Remarque : response_format est configuré au niveau de l'Agent ; chaque appel (y compris l'envoi d'e-mails,
# les demandes de suivi sans rapport avec "schedule summary") tenteront de force de générer un DailySummary,
# Il s'agit d'une limitation connue du response_format global. Pour créer véritablement un assistant multi-intentions, une meilleure approche consiste
# de faire de "generate daily summary" un outil séparé, en laissant l'Agent décider s'il faut l'appeler,
# plutôt que d'attacher un schéma de sortie toujours actif à l'ensemble de l'Agent.
if "structured_response" in final_state:
summary = final_state["structured_response"]
print("\n--- Résumé structuré (Markdown) ---)
print(to_markdown(summary))
return final_state
# ========== 7. Tests ==========
if __name__ == "__main__":
chat(Quel temps fait-il à Hangzhou aujourd'hui ? Consulte mon emploi du temps, puis donne-moi un résumé pour aujourd'hui)
chat(Aide-moi à envoyer un e-mail à [email protected], avec pour objet « Résumé quotidien » et un contenu indiquant que l'emploi du temps d'aujourd'hui a été confirmé)
# Troisième tour : aucune nouvelle information, on teste uniquement si l'Agent se souvient de la conversation précédente —
# Comme le même thread_id est transmis, le checkpointer ramènera l'historique complet
chat(Quel était l'objet de l'e-mail que je t'ai demandé d'envoyer tout à l'heure ?)
Résultat de l'exécution :
============================================================
You: 杭州今天天气怎么样？看看我的日程，然后给我一个今日总结
============================================================
🤔 决定调用工具: get_weather，参数: {'city': '杭州'}
🔧 get_weather 返回: 杭州天气：晴，温度 25°C，湿度 60%
🤔 决定调用工具: query_schedule，参数: {}
🔧 query_schedule 返回: 📅 2024-03-25 日程安排： - 10:00 团队周会（1小时） - 14:00...
🤖 助手: 早上好！今天杭州是大晴天，气温 25°C，湿度 60%，很适合出门活动。
今天您有两项日程：上午 10:00 的团队周会和下午 14:00 的 Python 学习，祝您今天顺利！
--- 结构化摘要（Markdown） ---
## 2024-03-25 今日摘要
- **天气**：杭州晴，25°C，湿度 60%
- **日程数量**：2
**重要事项：**
- 10:00 团队周会
- 14:00 Python 学习
**今日建议**：上午先参加团队周会，下午集中精力学习 Python，注意劳逸结合
============================================================
You: 帮我发一封邮件给 [email protected]，主题是'今日总结'，内容是今天日程已确认
============================================================
🤔 决定调用工具: send_email，参数: {'to': '[email protected]', 'subject': '今日总结', 'body': '今天日程已确认'}
🔧 send_email 返回: 邮件已发送！收件人：[email protected]，主题：今日总结，邮件ID：MSG-20240325143000
🤖 助手: 邮件已经发送成功啦！收件人是 [email protected]，主题"今日总结"。
--- 结构化摘要（Markdown） ---
## 2024-03-25 今日摘要
...（这里的摘要和邮件内容其实没什么关系，是 response_format 强制生成的，
属于前面提到的已知局限）
============================================================
You: 我刚才让你发的那封邮件，主题是什么来着？
============================================================
🤖 助手: 你刚才让我发的那封邮件主题是"今日总结"，收件人是 [email protected].
Le troisième tour n'a déclenché aucun appel d'outil du tout. L'assistant a pu répondre uniquement parce que
thread_idle même SqliteSaver a ramené l'historique complet des messages des deux premiers tours — voici ce à quoi ressemble vraiment la « mémoire de conversation » en action. Si vous le changezthread_idpour une nouvelle valeur, le troisième tour devient une conversation isolée, et l'assistant ne saura pas ce qui s'est passé « tout à l'heure ».
Résumé du projet
Cet assistant personnel démontre :
| Fonctionnalité | Implémentation |
|---|---|
| Collaboration multi-outils | Météo + emploi du temps + e-mail ; l'Agent choisit automatiquement l'ordre d'appel |
| Sortie structurée | Modèle Pydantic DailySummary + to_markdown() rendu sous forme de texte Markdown |
| Sortie en streaming | agent.stream(stream_mode="values"), affichant progressivement les appels d'outils et le processus de réflexion |
| Mémoire de conversation | SqliteSaver + thread_id, mémorisant le contexte sur plusieurs tours d'appels |
| Injection de date | @dynamic_prompt réécrit dynamiquement le prompt système, évitant la contamination de l'historique des messages |
| Interaction en langage naturel | Les utilisateurs décrivent leurs besoins en langage naturel, et l'Agent planifie de manière autonome |