LangChain Multi-Tool Personal Assistant
This article builds a personal assistant Agent that integrates weather query, schedule management, and email sending, demonstrating the complete usage of multi-tool collaboration and structured output.
System Design
- Three tools: weather query, schedule management, email sending
- Structured output: schedule summary formatted as Markdown
- Streaming output: real-time display of AI thinking and processing
- Conversation memory: remember user preferences and context
Complete Code
Before running, install dependencies, and.envconfigure it properly.DEEPSEEK_API_KEY：
pip install langchain langchain-deepseek langgraph-checkpoint-sqlite python-dotenv
Example
# File path: 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. Mock Data ==========
calendar_events = [
{"id": 1, "title": "Python Study", "date": "2024-03-25",
"time": "14:00", "duration": "2 hours"},
{"id": 2, "title": "Team Weekly Meeting", "date": "2024-03-25",
"time": "10:00", "duration": "1 hour"},
{"id": 3, "title": "Code Review", "date": "2024-03-26",
"time": "15:00", "duration": "1.5 hours"},
]
weather_db = {
"Hangzhou": {"condition": "Sunny", "temp": 25, "humidity": 60},
"Beijing": {"condition": "Cloudy", "temp": 18, "humidity": 45},
"Shanghai": {"condition": "Light Rain", "temp": 22, "humidity": 80},
}
# ========== 2. Define Tools ==========
@tool
def get_weather(city: str) -> str:
"""Query the real-time weather of a specified city.
Args:
city: City name, such as Hangzhou, Beijing, Shanghai
"""
data = weather_db.get(city)
if not data:
return f"Weather for {city} is not currently supported. Supported cities: {', '.join(weather_db.keys())}"
return (f"{city} weather: {data['condition']},"
f"Temperature {data['temp']}°C, humidity {data['humidity']}%")
@tool
def query_schedule(date: str = None) -> str:
"""Query the schedule for a specified date. If no date is specified, query today's schedule.
Args:
date: Date, format YYYY-MM-DD, e.g., 2024-03-25. If not provided, query today
"""
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} has no schedule."
events.sort(key=lambda e: e["time"])
# Here you must directly write the emoji character, not an HTML entity like 📅 —
# This is a Python string; it will be printed as-is and will not be parsed into a graphic by the browser
lines = [f"📅 {date} Schedule:"]
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:
"""Send email (simulated).
Args:
to: Recipient email
subject: Email subject
body: Email body
"""
# Simulate sending
email_id = f"MSG-{datetime.now().strftime('%Y%m%d%H%M%S')}"
return f"Email sent! Recipient: {to}, Subject: {subject}, Email ID: {email_id}"
# ========== 3. Structured Output Model ==========
class DailySummary(BaseModel):
"""Daily Summary"""
date: str = Field(description="Date")
weather_summary: str = Field(description="Weather Overview")
event_count: int = Field(description="Event Count")
key_events: list[str] = Field(description="Important Events List")
suggestion: str = Field(description="Today's Suggestion")
def to_markdown(summary: DailySummary) -> str:
"""Render the structured DailySummary as Markdown text,
corresponding to the "Structured output: schedule summary formatted as Markdown" item in the system design."""
lines = [
f"## {summary.date} Today's Summary",
"",
f"- **Weather**: {summary.weather_summary}",
f"- **Event Count**: {summary.event_count}",
"",
"**Important Items:**",
]
if summary.key_events:
lines += [f"- {event}" for event in summary.key_events]
else:
lines.append("- None")
lines += ["", f"**Today's Suggestion**: {summary.suggestion}"]
return "\n".join(lines)
# ========== 4. Define Middleware ==========
@dynamic_prompt
def inject_date_context(request) -> str:
"""Dynamically append current date information after the system prompt.
The previous approach of inserting the date message into the messages list with @before_model had two problems: 72. 1. The message update returned by before_model goes through the add_messages reducer, which only
1. The message updates returned by before_model go through the add_messages reducer; the reducer only according to
"appends new messages to the end", ignoring the position in the returned list; insert(-1, ...)
the intention to insert before the user message actually does not take effect;
2. before_model is triggered repeatedly in the multi-round tool-call loop, easily inserting this message
between AIMessage(tool_calls=...) and its corresponding ToolMessage,
disrupting the strict message order requirement.
Instead, use @dynamic_prompt to directly rewrite the system prompt string; every model call will
recompute it, without touching the messages list, so there is no risk of accumulation or misalignment.
"""
now = datetime.now()
weekday = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"][now.weekday()]
date_hint = (f"\n\n[System Prompt] Current date is {now.strftime('%Y-%m-%d')},
f"Weekday {weekday}. If the user does not specify a date, query today by default.")
return request.system_prompt + date_hint
# ========== 5. Create Agent ==========
# Use SqliteSaver to persist conversations, implementing the "conversation memory" in the system design.
# Establish the connection yourself and pass it to the SqliteSaver constructor, rather than using
# SqliteSaver.from_conn_string() (that is a context manager only suitable for with statements,
# a context manager that closes after use; see the article "LangChain Smart Customer Service Robot").
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="""You are a personal assistant named "Xiao Zhu". You can check the weather, manage schedules, and send emails.
## How it works
1. When the user asks "How is today?" or similar questions:
- First query today's weather (get_weather)
- Then query today's schedule (query_schedule)
- Then generate a daily summary
2. When the user asks to send an email, use the send_email tool
3. When the user only asks about weather or only asks about schedule, call only the corresponding tool
## Style
- Tone is friendly and natural
- Prefer using tools to get real-time data; do not make things up""",
)
# ========== 6. Interaction Function ==========
def chat(message: str, thread_id: str = "xiaoming"):
"""Chat with the assistant:
- Use stream_mode="values" to stream the AI's thinking and processing (corresponding to "streaming output")
- Pass in thread_id; conversations under the same thread_id will be remembered by SqliteSaver (corresponding to "conversation memory")
- If a structured summary is generated in this round, it will be rendered as Markdown and printed (corresponding to "structured output")
"""
config = {"configurable": {"thread_id": thread_id}}
print(f"\n{'='*60}")
print(f"You: {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", [])
# Each time a few more new messages appear, it means the Agent has taken a step forward; print them in real time
for msg in msgs[seen:]:
if msg.type == "ai" and getattr(msg, "tool_calls", None):
for call in msg.tool_calls:
print(f"🤔 Decided to call tool: {call['name']}, args: {call['args']}")
elif msg.type == "tool":
tool_name = getattr(msg, "name", "") or "Tool"
print(f"🔧 {tool_name} returned: {str(msg.content)[:80]}")
elif msg.type == "ai" and msg.content:
print(f"🤖 Assistant: {msg.content}")
seen = len(msgs)
# Note: response_format is configured at the Agent level; every call (including sending emails,
# follow-up requests unrelated to "schedule summary") will forcibly attempt to generate a DailySummary,
# This is a known limitation of the global response_format. To truly make a multi-intent assistant, a better approach is
# to make "generate daily summary" a separate tool, letting the Agent decide whether to call it,
# rather than attaching an always-active output schema to the entire Agent.
if "structured_response" in final_state:
summary = final_state["structured_response"]
print("\n--- Structured Summary (Markdown) ---)
print(to_markdown(summary))
return final_state
# ========== 7. Testing ==========
if __name__ == "__main__":
chat(What's the weather like in Hangzhou today? Check my schedule, then give me a summary for today)
chat(Help me send an email to [email protected], with the subject 'Daily Summary' and the content saying today's schedule has been confirmed)
# Third round: no new information, purely testing whether the Agent remembers the previous conversation —
# Because the same thread_id is passed, the checkpointer will bring back the full history
chat(What was the subject of the email I asked you to send just now?)
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. Mock Data ==========
calendar_events = [
{"id": 1, "title": "Python Study", "date": "2024-03-25",
"time": "14:00", "duration": "2 hours"},
{"id": 2, "title": "Team Weekly Meeting", "date": "2024-03-25",
"time": "10:00", "duration": "1 hour"},
{"id": 3, "title": "Code Review", "date": "2024-03-26",
"time": "15:00", "duration": "1.5 hours"},
]
weather_db = {
"Hangzhou": {"condition": "Sunny", "temp": 25, "humidity": 60},
"Beijing": {"condition": "Cloudy", "temp": 18, "humidity": 45},
"Shanghai": {"condition": "Light Rain", "temp": 22, "humidity": 80},
}
# ========== 2. Define Tools ==========
@tool
def get_weather(city: str) -> str:
"""Query the real-time weather of a specified city.
Args:
city: City name, such as Hangzhou, Beijing, Shanghai
"""
data = weather_db.get(city)
if not data:
return f"Weather for {city} is not currently supported. Supported cities: {', '.join(weather_db.keys())}"
return (f"{city} weather: {data['condition']},"
f"Temperature {data['temp']}°C, humidity {data['humidity']}%")
@tool
def query_schedule(date: str = None) -> str:
"""Query the schedule for a specified date. If no date is specified, query today's schedule.
Args:
date: Date, format YYYY-MM-DD, e.g., 2024-03-25. If not provided, query today
"""
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} has no schedule."
events.sort(key=lambda e: e["time"])
# Here you must directly write the emoji character, not an HTML entity like 📅 —
# This is a Python string; it will be printed as-is and will not be parsed into a graphic by the browser
lines = [f"📅 {date} Schedule:"]
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:
"""Send email (simulated).
Args:
to: Recipient email
subject: Email subject
body: Email body
"""
# Simulate sending
email_id = f"MSG-{datetime.now().strftime('%Y%m%d%H%M%S')}"
return f"Email sent! Recipient: {to}, Subject: {subject}, Email ID: {email_id}"
# ========== 3. Structured Output Model ==========
class DailySummary(BaseModel):
"""Daily Summary"""
date: str = Field(description="Date")
weather_summary: str = Field(description="Weather Overview")
event_count: int = Field(description="Event Count")
key_events: list[str] = Field(description="Important Events List")
suggestion: str = Field(description="Today's Suggestion")
def to_markdown(summary: DailySummary) -> str:
"""Render the structured DailySummary as Markdown text,
corresponding to the "Structured output: schedule summary formatted as Markdown" item in the system design."""
lines = [
f"## {summary.date} Today's Summary",
"",
f"- **Weather**: {summary.weather_summary}",
f"- **Event Count**: {summary.event_count}",
"",
"**Important Items:**",
]
if summary.key_events:
lines += [f"- {event}" for event in summary.key_events]
else:
lines.append("- None")
lines += ["", f"**Today's Suggestion**: {summary.suggestion}"]
return "\n".join(lines)
# ========== 4. Define Middleware ==========
@dynamic_prompt
def inject_date_context(request) -> str:
"""Dynamically append current date information after the system prompt.
The previous approach of inserting the date message into the messages list with @before_model had two problems: 72. 1. The message update returned by before_model goes through the add_messages reducer, which only
1. The message updates returned by before_model go through the add_messages reducer; the reducer only according to
"appends new messages to the end", ignoring the position in the returned list; insert(-1, ...)
the intention to insert before the user message actually does not take effect;
2. before_model is triggered repeatedly in the multi-round tool-call loop, easily inserting this message
between AIMessage(tool_calls=...) and its corresponding ToolMessage,
disrupting the strict message order requirement.
Instead, use @dynamic_prompt to directly rewrite the system prompt string; every model call will
recompute it, without touching the messages list, so there is no risk of accumulation or misalignment.
"""
now = datetime.now()
weekday = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"][now.weekday()]
date_hint = (f"\n\n[System Prompt] Current date is {now.strftime('%Y-%m-%d')},
f"Weekday {weekday}. If the user does not specify a date, query today by default.")
return request.system_prompt + date_hint
# ========== 5. Create Agent ==========
# Use SqliteSaver to persist conversations, implementing the "conversation memory" in the system design.
# Establish the connection yourself and pass it to the SqliteSaver constructor, rather than using
# SqliteSaver.from_conn_string() (that is a context manager only suitable for with statements,
# a context manager that closes after use; see the article "LangChain Smart Customer Service Robot").
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="""You are a personal assistant named "Xiao Zhu". You can check the weather, manage schedules, and send emails.
## How it works
1. When the user asks "How is today?" or similar questions:
- First query today's weather (get_weather)
- Then query today's schedule (query_schedule)
- Then generate a daily summary
2. When the user asks to send an email, use the send_email tool
3. When the user only asks about weather or only asks about schedule, call only the corresponding tool
## Style
- Tone is friendly and natural
- Prefer using tools to get real-time data; do not make things up""",
)
# ========== 6. Interaction Function ==========
def chat(message: str, thread_id: str = "xiaoming"):
"""Chat with the assistant:
- Use stream_mode="values" to stream the AI's thinking and processing (corresponding to "streaming output")
- Pass in thread_id; conversations under the same thread_id will be remembered by SqliteSaver (corresponding to "conversation memory")
- If a structured summary is generated in this round, it will be rendered as Markdown and printed (corresponding to "structured output")
"""
config = {"configurable": {"thread_id": thread_id}}
print(f"\n{'='*60}")
print(f"You: {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", [])
# Each time a few more new messages appear, it means the Agent has taken a step forward; print them in real time
for msg in msgs[seen:]:
if msg.type == "ai" and getattr(msg, "tool_calls", None):
for call in msg.tool_calls:
print(f"🤔 Decided to call tool: {call['name']}, args: {call['args']}")
elif msg.type == "tool":
tool_name = getattr(msg, "name", "") or "Tool"
print(f"🔧 {tool_name} returned: {str(msg.content)[:80]}")
elif msg.type == "ai" and msg.content:
print(f"🤖 Assistant: {msg.content}")
seen = len(msgs)
# Note: response_format is configured at the Agent level; every call (including sending emails,
# follow-up requests unrelated to "schedule summary") will forcibly attempt to generate a DailySummary,
# This is a known limitation of the global response_format. To truly make a multi-intent assistant, a better approach is
# to make "generate daily summary" a separate tool, letting the Agent decide whether to call it,
# rather than attaching an always-active output schema to the entire Agent.
if "structured_response" in final_state:
summary = final_state["structured_response"]
print("\n--- Structured Summary (Markdown) ---)
print(to_markdown(summary))
return final_state
# ========== 7. Testing ==========
if __name__ == "__main__":
chat(What's the weather like in Hangzhou today? Check my schedule, then give me a summary for today)
chat(Help me send an email to [email protected], with the subject 'Daily Summary' and the content saying today's schedule has been confirmed)
# Third round: no new information, purely testing whether the Agent remembers the previous conversation —
# Because the same thread_id is passed, the checkpointer will bring back the full history
chat(What was the subject of the email I asked you to send just now?)
Execution result:
============================================================
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]。
The third round did not trigger any tool calls at all. The assistant could answer purely because
thread_idthe same, SqliteSaver brought back the complete message history from the first two rounds — this is what "conversation memory" truly looks like in action. If you changethread_idit to a new value, the third round becomes an isolated conversation, and the assistant won't know what happened "just now".
Project Summary
This personal assistant demonstrates:
| Feature | Implementation |
|---|---|
| Multi-tool Collaboration | Weather + schedule + email; the Agent automatically chooses the calling order |
| Structured Output | DailySummary Pydantic model + to_markdown() rendered as Markdown text |
| Streaming Output | agent.stream(stream_mode="values"), progressively displaying tool calls and the thought process |
| Conversation Memory | SqliteSaver + thread_id, remembering context across multiple rounds of calls |
| Date Injection | @dynamic_prompt dynamically rewrites the system prompt, avoiding contamination of the message history |
| Natural Language Interaction | Users describe their needs in natural language, and the Agent plans autonomously |