TodoList 讓 AI 自動管理任務清單¶

from langchain_openai import AzureChatOpenAI
from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware
from langgraph.checkpoint.memory import InMemorySaver

# 創建帶有 TodoListMiddleware 的 agent
agent = create_agent(
    model=model,
    tools=[create_report, send_notification, schedule_meeting, analyze_data, send_email],
    system_prompt="""You are a helpful project manager assistant. Break down complex tasks into smaller steps and complete them systematically.

IMPORTANT: When working with multi-step tasks:
1. Create a todo list at the start
2. After completing EACH task, update the todo list to mark it as 'completed' and move the next task to 'in_progress'
3. Always use write_todos to update the status after each step""",
    middleware=[
        TodoListMiddleware(),  # 💡 無需任何參數配置！
    ],
    checkpointer=InMemorySaver(),  # 追蹤對話與任務狀態
)

對話流程：

用戶請求：「幫我準備 Q4 分析報告」

第 1 步：AI 自動呼叫 write_todos
┌─────────────────────────────────────────────────┐
│ 🔧 工具呼叫: write_todos                         │
│                                                 │
│ 參數:                                           │
│ {                                               │
│   'todos': [                                    │
│     {                                           │
│       'content': 'Analyze the Q4 sales data',   │
│       'status': 'in_progress'  ← 正在執行       │
│     },                                          │
│     {                                           │
│       'content': 'Create a summary report',     │
│       'status': 'pending'      ← 待辦           │
│     },                                          │
│     {                                           │
│       'content': 'Schedule a meeting',          │
│       'status': 'pending'      ← 待辦           │
│     },                                          │
│     {                                           │
│       'content': 'Send a notification',         │
│       'status': 'pending'      ← 待辦           │
│     }                                           │
│   ]                                             │
│ }                                               │
└─────────────────────────────────────────────────┘

第 2 步：AI 執行任務
• 按照清單順序執行每個任務
• 完成後更新狀態
• 確保不遺漏任何步驟

# 任務清單範例
todos = [
    {
        'content': 'Analyze the Q4 sales data to identify key insights and trends.',
        'status': 'in_progress'  # 目前正在做這個
    },
    {
        'content': 'Create a summary report with the findings from the Q4 sales data analysis.',
        'status': 'pending'  # 等待分析完成後執行
    },
    {
        'content': 'Schedule a meeting with the leadership team for tomorrow to present the Q4 analysis.',
        'status': 'pending'
    },
    {
        'content': 'Send a notification to all stakeholders about the upcoming Q4 analysis presentation.',
        'status': 'pending'
    }
]

# 使用 stream 來查看中間過程
for event in agent.stream(
    {"messages": [{"role": "user", "content": complex_task}]},
    config,
    stream_mode="updates"  # 💡 關鍵：使用 updates 模式
):
    if event:
        for node_name, node_update in event.items():
            print(f"→ 節點: {node_name}")

            # 檢查工具呼叫
            if "messages" in node_update:
                for msg in node_update["messages"]:
                    if hasattr(msg, 'tool_calls') and msg.tool_calls:
                        for tool_call in msg.tool_calls:
                            tool_name = tool_call.get('name', 'unknown')

                            # 偵測到 write_todos！
                            if 'todo' in tool_name.lower():
                                print(f"✓ TodoListMiddleware 正在管理任務！")
                                print(f"參數: {tool_call.get('args', {})}")

→ 節點: model
  🔧 呼叫工具: write_todos
     ✓ TodoListMiddleware 正在管理任務！
     參數: {'todos': [
         {'content': 'Analyze the Q4 sales data...', 'status': 'in_progress'},
         {'content': 'Create a summary report...', 'status': 'pending'},
         {'content': 'Schedule a meeting...', 'status': 'pending'},
         {'content': 'Send a notification...', 'status': 'pending'}
     ]}
→ 節點: tools
  內容: Updated todo list to [...]
→ 節點: model
  內容: Here is the plan for preparing your Q4 analysis presentation...

config = {"configurable": {"thread_id": "complex_task"}}

complex_task = """Please help me with the following tasks:
1. Analyze the Q4 sales data for the 'Electronics' category
2. Create a summary report titled 'Q4 Electronics Analysis'
3. Schedule a meeting with 'Leadership Team' on '2024-12-15' about 'Q4 Review'
4. Send a notification to 'All Stakeholders' about the meeting"""

# 使用 stream 觀察過程
for event in agent.stream(
    {"messages": [{"role": "user", "content": complex_task}]},
    config,
    stream_mode="updates"
):
    # ... 處理 stream 輸出，顯示每次狀態更新

📋 任務清單已建立:
  1. 🔄 in_progress: Analyze the Q4 sales data for the 'Electronics' category
  2. ⏳ pending: Create a summary report titled 'Q4 Electronics Analysis'
  3. ⏳ pending: Schedule a meeting with 'Leadership Team' on '2024-12-15' about 'Q4 Review'
  4. ⏳ pending: Send a notification to 'All Stakeholders' about the meeting

→ 節點: model
  🔧 呼叫工具: analyze_data
→ 節點: tools
  內容: 📊 Analysis complete for Q4 sales data - Electronics category: 15 insights found, 3 anomalies detected

📋 任務清單已建立:
  1. ✅ completed: Analyze the Q4 sales data for the 'Electronics' category  ← 已完成！
  2. 🔄 in_progress: Create a summary report titled 'Q4 Electronics Analysis'  ← 開始執行
  3. ⏳ pending: Schedule a meeting with 'Leadership Team' on '2024-12-15' about 'Q4 Review'
  4. ⏳ pending: Send a notification to 'All Stakeholders' about the meeting

→ 節點: model
  🔧 呼叫工具: create_report
→ 節點: tools
  內容: 📄 Report 'Q4 Electronics Analysis' created with 371 characters

📋 任務清單已建立:
  1. ✅ completed: Analyze the Q4 sales data for the 'Electronics' category
  2. ✅ completed: Create a summary report titled 'Q4 Electronics Analysis'  ← 已完成！
  3. 🔄 in_progress: Schedule a meeting with 'Leadership Team' on '2024-12-15' about 'Q4 Review'  ← 開始執行
  4. ⏳ pending: Send a notification to 'All Stakeholders' about the meeting

→ 節點: model
  🔧 呼叫工具: schedule_meeting
→ 節點: tools
  內容: 📅 Meeting scheduled for 2024-12-15 with Leadership Team about Q4 Review

📋 任務清單已建立:
  1. ✅ completed: Analyze the Q4 sales data for the 'Electronics' category
  2. ✅ completed: Create a summary report titled 'Q4 Electronics Analysis'
  3. ✅ completed: Schedule a meeting with 'Leadership Team' on '2024-12-15' about 'Q4 Review'  ← 已完成！
  4. 🔄 in_progress: Send a notification to 'All Stakeholders' about the meeting  ← 開始執行

→ 節點: model
  🔧 呼叫工具: send_notification
→ 節點: tools
  內容: 📬 Notification sent to All Stakeholders: A meeting regarding the Q4 Review is scheduled...

📋 任務清單已建立:
  1. ✅ completed: Analyze the Q4 sales data for the 'Electronics' category
  2. ✅ completed: Create a summary report titled 'Q4 Electronics Analysis'
  3. ✅ completed: Schedule a meeting with 'Leadership Team' on '2024-12-15' about 'Q4 Review'
  4. ✅ completed: Send a notification to 'All Stakeholders' about the meeting  ← 全部完成！

All tasks have been completed:

1. Q4 sales data for the 'Electronics' category was analyzed (15 insights, 3 anomalies found).
2. A summary report titled 'Q4 Electronics Analysis' was created.
3. A meeting with the Leadership Team about 'Q4 Review' was scheduled for 2024-12-15.
4. All Stakeholders were notified about the meeting.

If you need the report details or want to take further actions, please let me know!

# 基本版本
agent = create_agent(
    model=model,
    tools=[...],
    middleware=[TodoListMiddleware()],
    checkpointer=InMemorySaver(),
)

# 優化版本 ✅
agent = create_agent(
    model=model,
    tools=[...],
    system_prompt="""You are a helpful project manager assistant. 
    Break down complex tasks into smaller steps and complete them systematically.

    When given a multi-step task:
    1. Create a detailed todo list with clear steps
    2. Mark the current step as 'in_progress'
    3. Execute tasks in logical order
    4. Update task status as you progress
    """,
    middleware=[TodoListMiddleware()],
    checkpointer=InMemorySaver(),
)

from langchain.agents.middleware import (
    TodoListMiddleware,
    HumanInTheLoopMiddleware,
    SummarizationMiddleware
)

# 組合多個 Middleware
agent = create_agent(
    model=model,
    tools=[...],
    middleware=[
        TodoListMiddleware(),  # 任務管理
        HumanInTheLoopMiddleware(  # 重要步驟需要人工確認
            approve_all_tools=False,
            tools_requiring_approval=["send_email", "schedule_meeting"]
        ),
        SummarizationMiddleware(  # 長對話自動摘要
            model=model,
            max_tokens_before_summary=1000,
            messages_to_keep=5,
        ),
    ],
    checkpointer=InMemorySaver(),
)

def monitor_todo_activity(result):
    """監控任務清單活動"""
    todo_calls = 0

    for msg in result['messages']:
        if hasattr(msg, 'tool_calls') and msg.tool_calls:
            for call in msg.tool_calls:
                if 'todo' in call.get('name', '').lower():
                    todo_calls += 1
                    print(f"發現任務清單操作：{call.get('name')}")
                    print(f"參數：{call.get('args', {})}")

    return todo_calls

# 使用
result = agent.invoke({"messages": [{"role": "user", "content": task}]}, config)
activity_count = monitor_todo_activity(result)
print(f"總共進行了 {activity_count} 次任務清單操作")

# 執行任務
config = {"configurable": {"thread_id": "my_task"}}
agent.invoke({"messages": [{"role": "user", "content": complex_task}]}, config)

# 讀取當前狀態
state = agent.get_state(config)

# 分析任務清單
messages = state.values.get("messages", [])
for msg in messages:
    if hasattr(msg, 'tool_calls') and msg.tool_calls:
        for call in msg.tool_calls:
            if 'todo' in call.get('name', '').lower():
                todos = call.get('args', {}).get('todos', [])

                print("當前任務清單：")
                for i, todo in enumerate(todos, 1):
                    status_icon = {
                        'pending': '⏳',
                        'in_progress': '🔄',
                        'completed': '✅'
                    }.get(todo.get('status', 'pending'), '❓')

                    print(f"{i}. {status_icon} {todo.get('content')} ({todo.get('status')})")

當前任務清單：
1. 🔄 Analyze the Q4 sales data to identify key insights and trends. (in_progress)
2. ⏳ Create a summary report with the findings from the Q4 sales data analysis. (pending)
3. ⏳ Schedule a meeting with the leadership team for tomorrow to present the Q4 analysis. (pending)
4. ⏳ Send a notification to all stakeholders about the upcoming Q4 analysis presentation. (pending)

def track_task_completion(agent, config, task_description):
    """追蹤任務完成情況"""

    # 執行任務
    result = agent.invoke(
        {"messages": [{"role": "user", "content": task_description}]},
        config
    )

    # 分析完成情況
    todos = []
    for msg in result['messages']:
        if hasattr(msg, 'tool_calls') and msg.tool_calls:
            for call in msg.tool_calls:
                if 'todo' in call.get('name', '').lower():
                    todos = call.get('args', {}).get('todos', [])

    if todos:
        total = len(todos)
        completed = sum(1 for t in todos if t.get('status') == 'completed')
        in_progress = sum(1 for t in todos if t.get('status') == 'in_progress')
        pending = sum(1 for t in todos if t.get('status') == 'pending')

        print(f"任務進度：{completed}/{total} 已完成")
        print(f"進行中：{in_progress} | 待辦：{pending}")

        return {
            'total': total,
            'completed': completed,
            'in_progress': in_progress,
            'pending': pending,
            'completion_rate': (completed / total * 100) if total > 0 else 0
        }

    return None

# 使用
stats = track_task_completion(agent, config, "Prepare Q4 analysis presentation")
if stats:
    print(f"完成率：{stats['completion_rate']:.1f}%")

# 簡單任務：不建立清單
"Send a notification to the team"
→ 直接執行 send_notification ✓
→ write_todos 呼叫：0 次

# 複雜任務：自動建立清單
"Prepare Q4 analysis presentation: analyze data, create report, schedule meeting, send notification"
→ 呼叫 write_todos 建立 4 個子任務 ✓
→ write_todos 呼叫：1 次

# 基本配置：可能不會自動更新
agent = create_agent(
    model=model,
    tools=[...],
    middleware=[TodoListMiddleware()],
)

# 優化配置：加入狀態更新指示 ✅
agent = create_agent(
    model=model,
    tools=[...],
    system_prompt="""You are a project manager assistant.
    When completing tasks:
    1. Update task status from 'pending' to 'in_progress' when starting
    2. Update to 'completed' when finished
    3. Always call write_todos to update the list
    """,
    middleware=[TodoListMiddleware()],
)

system_prompt = """You are a disciplined project manager assistant.

STRICT RULES:
1. ALWAYS create a todo list for multi-step tasks
2. ONLY execute tasks in the order listed
3. COMPLETE the current task before moving to the next
4. UPDATE task status after each action
5. NEVER skip tasks unless explicitly told to do so

Task status flow:
pending → in_progress → completed
"""

agent = create_agent(
    model=model,
    tools=[...],
    system_prompt=system_prompt,
    middleware=[TodoListMiddleware()],
)

# 完整的企業級 Agent 配置
agent = create_agent(
    model=model,
    tools=enterprise_tools,
    middleware=[
        # 1. 任務管理（最底層）
        TodoListMiddleware(),

        # 2. 人工審批（中間層）
        HumanInTheLoopMiddleware(
            approve_all_tools=False,
            tools_requiring_approval=["send_email", "delete_data", "schedule_meeting"]
        ),

        # 3. 對話壓縮（最上層）
        SummarizationMiddleware(
            model=summary_model,
            max_tokens_before_summary=1000,
            messages_to_keep=5,
        ),
    ],
    checkpointer=PostgresSaver(...),
)