LangChain: персональный ассистент с несколькими инструментами
В этой статье создаётся агент персонального ассистента, который объединяет запрос погоды, управление расписанием и отправку электронной почты, демонстрируя полное использование совместной работы нескольких инструментов и структурированного вывода.
Проектирование систем
- Три инструмента: запрос погоды, управление расписанием, отправка электронной почты
- Структурированный вывод: сводка расписания в формате Markdown
- Потоковый вывод: отображение мыслей и обработки ИИ в реальном времени
- Память диалога: запоминание предпочтений и контекста пользователя
Полный код
Перед запуском установите зависимости и.envправильно настройте его.DEEPSEEK_API_KEY：
pip install langchain langchain-deepseek langgraph-checkpoint-sqlite python-dotenv
Пример
# Путь к файлу: 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. Тестовые данные ==========
calendar_events = [
{"id": 1, "title": "Изучение Python", "date": "2024-03-25",
"time": "14:00", "duration": "2 часа"},
{"id": 2, "title": "Еженедельная встреча команды", "date": "2024-03-25",
"time": "10:00", "duration": "1 час"},
{"id": 3, "title": "Ревью кода", "date": "2024-03-26",
"time": "15:00", "duration": "1.5 часа"},
]
weather_db = {
"Ханчжоу": {"condition": "Солнечно", "temp": 25, "humidity": 60},
"Пекин": {"condition": "Облачно", "temp": 18, "humidity": 45},
"Шанхай": {"condition": "Небольшой дождь", "temp": 22, "humidity": 80},
}
# ========== 2. Определение инструментов ==========
@tool
def get_weather(city: str) -> str:
"""Запросить погоду в реальном времени для указанного города.
Аргументы:
city: Название города, например Ханчжоу, Пекин, Шанхай
"""
data = weather_db.get(city)
if not data:
return f"Погода для {city} в настоящее время не поддерживается. Поддерживаемые города: {', '.join(weather_db.keys())}"
return (f"Погода в {city}: {data['condition']},"
f"Температура {data['temp']}°C, влажность {data['humidity']}%")
@tool
def query_schedule(date: str = None) -> str:
"""Запросить расписание на указанную дату. Если дата не указана, запросить расписание на сегодня.
Аргументы:
date: Date, формат YYYY-MM-DD, например, 2024-03-25. Если не указано, запросить расписание на сегодня
"""
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} нет расписания."
events.sort(key=lambda e: e["time"])
# Здесь вы должны напрямую написать символ эмодзи, а не HTML-сущность вроде 📅 —
# Это строка Python; она будет выведена как есть и не будет преобразована браузером в графическое изображение
lines = [f"📅 {date} Расписание:"]
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:
"""Отправить электронное письмо (имитация).
Аргументы:
to: электронная почта получателя
subject: тема письма
body: тело письма
"""
# Имитировать отправку
email_id = f"MSG-{datetime.now().strftime('%Y%m%d%H%M%S')}"
return f"Письмо отправлено! Получатель: {to}, Тема: {subject}, ID письма: {email_id}"
# ========== 3. Модель структурированного вывода ==========
class DailySummary(BaseModel):
"""Ежедневная сводка"""
date: str = Field(description="Дата")
weather_summary: str = Field(description="Обзор погоды")
event_count: int = Field(description="Количество событий")
key_events: list[str] = Field(description="Список важных событий")
suggestion: str = Field(description="Совет на сегодня")
def to_markdown(summary: DailySummary) -> str:
"""Преобразует структурированный DailySummary в текст Markdown,
соответствующий пункту "Структурированный вывод: сводка расписания в формате Markdown" в проекте системы."""
lines = [
f"## {summary.date} Сводка за сегодня",
"",
f"- **Погода**: {summary.weather_summary}",
f"- **Количество событий**: {summary.event_count}",
"",
"**Важные пункты:**",
]
if summary.key_events:
lines += [f"- {event}" for event in summary.key_events]
else:
lines.append("- Нет")
lines += ["", f"**Рекомендация на сегодня**: {summary.suggestion}"]
return "\n".join(lines)
# ========== 4. Определить Middleware ==========
@dynamic_prompt
def inject_date_context(request) -> str:
"""Динамически добавляет информацию о текущей дате после системного промпта.
Предыдущий подход с вставкой сообщения с датой в список messages с помощью @before_model имел две проблемы: 72. 1. Обновление сообщения, возвращаемое before_model, проходит через редьюсер add_messages, который только
1. Обновления сообщений, возвращаемые before_model, проходят через редьюсер add_messages; редьюсер только в соответствии с
"добавляет новые сообщения в конец", игнорируя позицию в возвращаемом списке; insert(-1, ...)
намерение вставить перед сообщением пользователя фактически не даёт эффекта;
2. before_model срабатывает многократно в цикле многораундовых вызовов инструментов, легко вставляя это сообщение
между AIMessage(tool_calls=...) и соответствующим ему ToolMessage,
нарушая требование строгого порядка сообщений.
Вместо этого используйте @dynamic_prompt, чтобы напрямую перезаписывать строку системного промпта; каждый вызов модели будет
пересчитывать её, не затрагивая список messages, поэтому нет риска накопления или смещения.
"""
now = datetime.now()
weekday = ["Пн", "Вт", "Ср", "Чт", "Пт", "Сб", "Вс"][now.weekday()]
date_hint = (f"\n\n[Системный промпт] Текущая дата: {now.strftime('%Y-%m-%d')},
f"День недели {weekday}. Если пользователь не указывает дату, по умолчанию запрашивается сегодняшняя.")
return request.system_prompt + date_hint
# ========== 5. Создание агента ==========
# Используйте SqliteSaver для сохранения диалогов, реализуя «память диалога» в системном дизайне.
# Установите соединение самостоятельно и передайте его в конструктор SqliteSaver, вместо использования
# SqliteSaver.from_conn_string() (это контекстный менеджер, подходящий только для операторов with,
# контекстный менеджер, который закрывается после использования; см. статью «Умный робот службы поддержки клиентов 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_format=DailySummary,
checkpointer=checkpointer,
system_prompt="""Ты — персональный ассистент по имени "Сяо Чжу". Ты можешь проверять погоду, управлять расписанием и отправлять электронные письма.
## Как это работает
1. Когда пользователь спрашивает "Как сегодня?" или задаёт похожие вопросы:
- Сначала запросите сегодняшнюю погоду (get_weather)
- Затем запросите расписание на сегодня (query_schedule)
- Затем сформируйте сводку за день
2. Когда пользователь просит отправить электронное письмо, используйте инструмент send_email
3. Когда пользователь спрашивает только о погоде или только о расписании, вызывайте только соответствующий инструмент
## Стиль
- Тон дружелюбный и естественный
- Предпочитайте использовать инструменты для получения данных в реальном времени; не выдумывайте""",
)
# ========== 6. Функция взаимодействия ==========
def chat(message: str, thread_id: str = "xiaoming"):
"""Чат с ассистентом:
- Используйте stream_mode="values" для потоковой передачи мыслей и обработки ИИ (соответствует "потоковому выводу")
- Передайте thread_id; разговоры с одинаковым thread_id будут запоминаться SqliteSaver (соответствует "памяти диалога")
- Если в этом раунде будет создана структурированная сводка, она будет отображена как Markdown и выведена на печать (соответствует "структурированному выводу")
"""
config = {"configurable": {"thread_id": thread_id}}
print(f"\n{'='*60}")
print(f"Вы: {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", [])
# Каждый раз, когда появляется ещё несколько новых сообщений, это означает, что Агент сделал шаг вперёд; выводите их в реальном времени
for msg in msgs[seen:]:
if msg.type == "ai" and getattr(msg, "tool_calls", None):
for call in msg.tool_calls:
print(f"🤔 Решено вызвать инструмент: {call['name']}, аргументы: {call['args']}")
elif msg.type == "tool":
tool_name = getattr(msg, "name", "") or "Инструмент"
print(f"🔧 {tool_name} вернул: {str(msg.content)[:80]}")
elif msg.type == "ai" and msg.content:
print(f"🤖 Ассистент: {msg.content}")
seen = len(msgs)
# Примечание: response_format настроен на уровне Agent; каждый вызов (включая отправку электронных писем,
# последующие запросы, не связанные со "сводкой расписания"), будут принудительно пытаться сгенерировать DailySummary,
# Это известное ограничение глобального response_format. Чтобы действительно сделать мультиинтентного ассистента, лучший подход —
# сделать "генерацию ежедневной сводки" отдельным инструментом, позволяя Agent решать, вызывать ли его,
# а не прикреплять всегда активную выходную схему ко всему Agent.
if "structured_response" in final_state:
summary = final_state["structured_response"]
print("\n--- Структурированная сводка (Markdown) ---)
print(to_markdown(summary))
return final_state
# ========== 7. Тестирование ==========
if __name__ == "__main__":
chat(Какая сегодня погода в Ханчжоу? Проверь моё расписание, затем дай мне сводку на сегодня)
chat(Помоги мне отправить письмо на [email protected] с темой 'Ежедневная сводка' и содержанием, в котором говорится, что сегодняшнее расписание подтверждено)
# Третий раунд: новой информации нет, чисто проверяем, помнит ли Агент предыдущий разговор —
# Поскольку передаётся тот же thread_id, checkpointer вернёт полную историю
chat(Какая была тема письма, которое я только что просил тебя отправить?)
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. Тестовые данные ==========
calendar_events = [
{"id": 1, "title": "Изучение Python", "date": "2024-03-25",
"time": "14:00", "duration": "2 часа"},
{"id": 2, "title": "Еженедельная встреча команды", "date": "2024-03-25",
"time": "10:00", "duration": "1 час"},
{"id": 3, "title": "Ревью кода", "date": "2024-03-26",
"time": "15:00", "duration": "1.5 часа"},
]
weather_db = {
"Ханчжоу": {"condition": "Солнечно", "temp": 25, "humidity": 60},
"Пекин": {"condition": "Облачно", "temp": 18, "humidity": 45},
"Шанхай": {"condition": "Небольшой дождь", "temp": 22, "humidity": 80},
}
# ========== 2. Определение инструментов ==========
@tool
def get_weather(city: str) -> str:
"""Запросить погоду в реальном времени для указанного города.
Аргументы:
city: Название города, например Ханчжоу, Пекин, Шанхай
"""
data = weather_db.get(city)
if not data:
return f"Погода для {city} в настоящее время не поддерживается. Поддерживаемые города: {', '.join(weather_db.keys())}"
return (f"Погода в {city}: {data['condition']},"
f"Температура {data['temp']}°C, влажность {data['humidity']}%")
@tool
def query_schedule(date: str = None) -> str:
"""Запросить расписание на указанную дату. Если дата не указана, запросить расписание на сегодня.
Аргументы:
date: Date, формат YYYY-MM-DD, например, 2024-03-25. Если не указано, запросить расписание на сегодня
"""
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} нет расписания."
events.sort(key=lambda e: e["time"])
# Здесь вы должны напрямую написать символ эмодзи, а не HTML-сущность вроде 📅 —
# Это строка Python; она будет выведена как есть и не будет преобразована браузером в графическое изображение
lines = [f"📅 {date} Расписание:"]
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:
"""Отправить электронное письмо (имитация).
Аргументы:
to: электронная почта получателя
subject: тема письма
body: тело письма
"""
# Имитировать отправку
email_id = f"MSG-{datetime.now().strftime('%Y%m%d%H%M%S')}"
return f"Письмо отправлено! Получатель: {to}, Тема: {subject}, ID письма: {email_id}"
# ========== 3. Модель структурированного вывода ==========
class DailySummary(BaseModel):
"""Ежедневная сводка"""
date: str = Field(description="Дата")
weather_summary: str = Field(description="Обзор погоды")
event_count: int = Field(description="Количество событий")
key_events: list[str] = Field(description="Список важных событий")
suggestion: str = Field(description="Совет на сегодня")
def to_markdown(summary: DailySummary) -> str:
"""Преобразует структурированный DailySummary в текст Markdown,
соответствующий пункту "Структурированный вывод: сводка расписания в формате Markdown" в проекте системы."""
lines = [
f"## {summary.date} Сводка за сегодня",
"",
f"- **Погода**: {summary.weather_summary}",
f"- **Количество событий**: {summary.event_count}",
"",
"**Важные пункты:**",
]
if summary.key_events:
lines += [f"- {event}" for event in summary.key_events]
else:
lines.append("- Нет")
lines += ["", f"**Рекомендация на сегодня**: {summary.suggestion}"]
return "\n".join(lines)
# ========== 4. Определить Middleware ==========
@dynamic_prompt
def inject_date_context(request) -> str:
"""Динамически добавляет информацию о текущей дате после системного промпта.
Предыдущий подход с вставкой сообщения с датой в список messages с помощью @before_model имел две проблемы: 72. 1. Обновление сообщения, возвращаемое before_model, проходит через редьюсер add_messages, который только
1. Обновления сообщений, возвращаемые before_model, проходят через редьюсер add_messages; редьюсер только в соответствии с
"добавляет новые сообщения в конец", игнорируя позицию в возвращаемом списке; insert(-1, ...)
намерение вставить перед сообщением пользователя фактически не даёт эффекта;
2. before_model срабатывает многократно в цикле многораундовых вызовов инструментов, легко вставляя это сообщение
между AIMessage(tool_calls=...) и соответствующим ему ToolMessage,
нарушая требование строгого порядка сообщений.
Вместо этого используйте @dynamic_prompt, чтобы напрямую перезаписывать строку системного промпта; каждый вызов модели будет
пересчитывать её, не затрагивая список messages, поэтому нет риска накопления или смещения.
"""
now = datetime.now()
weekday = ["Пн", "Вт", "Ср", "Чт", "Пт", "Сб", "Вс"][now.weekday()]
date_hint = (f"\n\n[Системный промпт] Текущая дата: {now.strftime('%Y-%m-%d')},
f"День недели {weekday}. Если пользователь не указывает дату, по умолчанию запрашивается сегодняшняя.")
return request.system_prompt + date_hint
# ========== 5. Создание агента ==========
# Используйте SqliteSaver для сохранения диалогов, реализуя «память диалога» в системном дизайне.
# Установите соединение самостоятельно и передайте его в конструктор SqliteSaver, вместо использования
# SqliteSaver.from_conn_string() (это контекстный менеджер, подходящий только для операторов with,
# контекстный менеджер, который закрывается после использования; см. статью «Умный робот службы поддержки клиентов 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_format=DailySummary,
checkpointer=checkpointer,
system_prompt="""Ты — персональный ассистент по имени "Сяо Чжу". Ты можешь проверять погоду, управлять расписанием и отправлять электронные письма.
## Как это работает
1. Когда пользователь спрашивает "Как сегодня?" или задаёт похожие вопросы:
- Сначала запросите сегодняшнюю погоду (get_weather)
- Затем запросите расписание на сегодня (query_schedule)
- Затем сформируйте сводку за день
2. Когда пользователь просит отправить электронное письмо, используйте инструмент send_email
3. Когда пользователь спрашивает только о погоде или только о расписании, вызывайте только соответствующий инструмент
## Стиль
- Тон дружелюбный и естественный
- Предпочитайте использовать инструменты для получения данных в реальном времени; не выдумывайте""",
)
# ========== 6. Функция взаимодействия ==========
def chat(message: str, thread_id: str = "xiaoming"):
"""Чат с ассистентом:
- Используйте stream_mode="values" для потоковой передачи мыслей и обработки ИИ (соответствует "потоковому выводу")
- Передайте thread_id; разговоры с одинаковым thread_id будут запоминаться SqliteSaver (соответствует "памяти диалога")
- Если в этом раунде будет создана структурированная сводка, она будет отображена как Markdown и выведена на печать (соответствует "структурированному выводу")
"""
config = {"configurable": {"thread_id": thread_id}}
print(f"\n{'='*60}")
print(f"Вы: {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", [])
# Каждый раз, когда появляется ещё несколько новых сообщений, это означает, что Агент сделал шаг вперёд; выводите их в реальном времени
for msg in msgs[seen:]:
if msg.type == "ai" and getattr(msg, "tool_calls", None):
for call in msg.tool_calls:
print(f"🤔 Решено вызвать инструмент: {call['name']}, аргументы: {call['args']}")
elif msg.type == "tool":
tool_name = getattr(msg, "name", "") or "Инструмент"
print(f"🔧 {tool_name} вернул: {str(msg.content)[:80]}")
elif msg.type == "ai" and msg.content:
print(f"🤖 Ассистент: {msg.content}")
seen = len(msgs)
# Примечание: response_format настроен на уровне Agent; каждый вызов (включая отправку электронных писем,
# последующие запросы, не связанные со "сводкой расписания"), будут принудительно пытаться сгенерировать DailySummary,
# Это известное ограничение глобального response_format. Чтобы действительно сделать мультиинтентного ассистента, лучший подход —
# сделать "генерацию ежедневной сводки" отдельным инструментом, позволяя Agent решать, вызывать ли его,
# а не прикреплять всегда активную выходную схему ко всему Agent.
if "structured_response" in final_state:
summary = final_state["structured_response"]
print("\n--- Структурированная сводка (Markdown) ---)
print(to_markdown(summary))
return final_state
# ========== 7. Тестирование ==========
if __name__ == "__main__":
chat(Какая сегодня погода в Ханчжоу? Проверь моё расписание, затем дай мне сводку на сегодня)
chat(Помоги мне отправить письмо на [email protected] с темой 'Ежедневная сводка' и содержанием, в котором говорится, что сегодняшнее расписание подтверждено)
# Третий раунд: новой информации нет, чисто проверяем, помнит ли Агент предыдущий разговор —
# Поскольку передаётся тот же thread_id, checkpointer вернёт полную историю
chat(Какая была тема письма, которое я только что просил тебя отправить?)
Результат выполнения:
============================================================
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].
Третий раунд вообще не вызвал никаких вызовов инструментов. Ассистент смог ответить только потому, что
thread_idтот же, SqliteSaver вернул полную историю сообщений из первых двух раундов — вот как «память разговора» на самом деле выглядит в действии. Если вы изменитеthread_idего на новое значение, третий раунд становится изолированным разговором, и ассистент не будет знать, что произошло «только что».
Сводка проекта
Этот персональный ассистент демонстрирует:
| Характеристика | Реализация |
|---|---|
| Совместная работа нескольких инструментов | Погода + расписание + электронная почта; Агент автоматически выбирает порядок вызовов |
| Структурированный вывод | Модель Pydantic DailySummary + to_markdown(), отрендеренная как текст Markdown |
| Потоковый вывод | agent.stream(stream_mode="values"), постепенно отображая вызовы инструментов и процесс мышления |
| Память диалога | SqliteSaver + thread_id, запоминание контекста между несколькими раундами вызовов |
| Внедрение даты | @dynamic_prompt динамически перезаписывает системный промпт, избегая загрязнения истории сообщений |
| Взаимодействие на естественном языке | Пользователи описывают свои потребности на естественном языке, а Агент планирует автономно |