LangChain マルチツール個人アシスタント

# ファイルパス：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:
    """指定都市のリアルタイム天気を照会します。

    Args:
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:
    """指定日のスケジュールを照会します。日付を指定しない場合は今日のスケジュールを照会します。

    Args:
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"])
    # ここでは emoji 文字を直接書く必要があり、📅 のような 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:
    """メールを送信します（シミュレーション）。

    Args:
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:
    """システムプロンプトの末尾に現在の日付情報を動的に追加します。

以前、@before_model で日付メッセージを messages リストに挿入する書き方には2つの問題がありました：
1. before_model が返すメッセージの更新は add_messages reducer を経由しますが、reducer は単に
"新メッセージを末尾に追加"という処理だけを行い、返されたリスト内の位置は認識しません。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. Agent を作成 ==========

# 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" を使って AI の思考と処理過程をストリーミング表示する（「ストリーミング出力」に対応）
- 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", [])
        # 新しいメッセージがいくつか増えるたびに、Agent が一歩進んだことを示すので、リアルタイムに出力する
        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] にメールを送ってください。件名は「今日のサマリー」、内容は今日のスケジュールが確認済みです")
    # 第3ラウンドは新しい情報を一切含まず、純粋にエージェントが前のラウンドの会話を覚えているかを試す——
    # 同じ thread_id を渡しているため、チェックポインターが完全な履歴を持ち帰る
    chat("さっき送ってもらったあのメールの件名は何でしたっけ？")