Chapter 5: LangGraph 狀態機 (Graph Workflows)

歡迎來到 open-notebook 教學系列的第五章！在上一章 內容處理流程 (Content Processing Graph) 中，我們了解到 open-notebook 如何使用一個自動化的流程來從各種來源（如網址、PDF、音檔）提取純文字內容。我們也稍微提到了這個流程是使用 LangGraph 這個函式庫建立的。現在，我們要更深入地探索 LangGraph，看看它是如何讓我們能夠編排更複雜、多步驟的人工智慧任務。

為什麼我們需要 LangGraph？

想像一下，您想在 open-notebook 中加入一個「智慧問答」功能。您希望能直接向您的筆記提問，例如：「我上次關於 AI 倫理的筆記中，提到了哪些主要觀點？」

要回答這個問題，單純呼叫一次 AI 模型可能不夠。一個比較完善的處理流程可能會是這樣：

  1. 理解與規劃： 首先，一個 AI 模型（或一個工具）需要理解您的問題，並規劃出一個「搜尋策略」。例如，它可能會決定要搜尋「AI 倫理」和「主要觀點」這兩個關鍵詞，並且指示後續步驟要從搜尋結果中提取相關的論點。
  2. 執行搜尋： 接著，系統需要在您的筆記資料庫中執行搜尋。這可能不只一次，而是根據策略進行多次不同角度的搜尋。
  3. 結果處理與回答： 拿到搜尋結果後，另一個 AI 模型需要閱讀這些結果，並結合您原始的問題，生成一個完整且易於理解的回答。
  4. 決策與循環（可能）： 如果第一次搜尋結果不理想，系統是否應該自動調整策略，嘗試不同的關鍵詞，然後重新搜尋呢？這就涉及到條件判斷和循環。

這種包含多個步驟、條件分支甚至循環的複雜任務，如果只用傳統的程式碼一行行寫下來，很快就會變得難以管理和維護。每次想調整流程中的某個環節，都可能牽一髮而動全身。

這就是 LangGraph 發揮作用的地方。LangGraph 就像一位電影導演，手上有個劇本（工作流程圖）。劇本中定義了不同的場景（節點）和演員（AI 模型、工具）的出場順序（邊）。導演會根據劇本，一步步指導演員完成拍攝，確保整個流程順暢地進行，最終完成一部電影（任務）。

什麼是 LangGraph？

LangGraph 是一個 Python 函式庫，專門用來建構狀態化、多參與者的應用程式，特別適合需要協調多個大型語言模型 (LLM) 和工具來完成複雜任務的場景。它讓您可以將複雜的流程定義成一個「圖」(Graph)。

以下是 LangGraph 的幾個核心概念，讓我們用電影導演的例子來理解：

在 open-notebook 中，內容處理流程 (Content Processing Graph) 就是用 LangGraph 建立的。除此之外，「智慧問答」(Ask)、「聊天」(Chat)、「處理新來源並應用轉換」(Source processing with Transformations) 等功能，也都是透過 LangGraph 編排的複雜工作流程。

LangGraph 如何在 open-notebook 中運作：「智慧問答」範例

讓我們以 open-notebook 中的「智慧問答」(Ask) 功能為例，看看 LangGraph 是如何運作的。這個功能的程式碼主要位於 open_notebook/graphs/ask.py。

當您問：「我關於 AI 倫理的筆記中，提到了哪些主要觀點？」時，背後的 LangGraph 工作流程大致如下：

1. 定義狀態 (State)

首先，我們需要定義在這個「智慧問答」流程中，需要在節點之間傳遞哪些資訊。這通常透過一個 Python 的 TypedDict 來定義。

# open_notebook/graphs/ask.py (簡化片段)
from typing import Annotated, List
from pydantic import BaseModel, Field
from typing_extensions import TypedDict
import operator # 用於合併列表

class Search(BaseModel): # 代表一次搜尋請求的結構
    term: str # 搜尋詞彙
    instructions: str # 給回答模型的指示，告訴它要從搜尋結果中提取什麼資訊

class Strategy(BaseModel): # 代表搜尋策略的結構
    reasoning: str # AI 思考為何要這樣搜尋的理由
    searches: List[Search] = Field(default_factory=list) # 包含多次搜尋請求的列表

class ThreadState(TypedDict): # 定義整個流程的狀態
    question: str               # 使用者提出的原始問題
    strategy: Strategy          # AI 生成的搜尋策略
    answers: Annotated[list, operator.add] # 從各個搜尋中收集到的答案片段 (會不斷累加)
    final_answer: str           # 最終給使用者的完整回答

程式碼解釋：

2. 定義節點 (Nodes)

接下來，我們定義流程中的各個「演員」（處理步驟）。每個節點都是一個 Python 函式，它會接收目前的 ThreadState，執行一些操作，然後返回一個字典來更新狀態。

open_notebook/graphs/ask.py 中定義了幾個主要節點：

3. 定義邊 (Edges)

定義好節點後，我們需要用「邊」來連接它們，告訴 LangGraph 流程應該如何進行。

# open_notebook/graphs/ask.py (簡化片段)
from langgraph.graph import StateGraph, END, START, Send

# 假設 agent_state 是 StateGraph(ThreadState) 的實例
agent_state = StateGraph(ThreadState)

# 新增節點 (前面已定義)
agent_state.add_node("agent", call_model_with_messages) # "agent" 是策略制定節點的名稱
agent_state.add_node("provide_answer", provide_answer)
agent_state.add_node("write_final_answer", write_final_answer)

# 設定起始點
agent_state.add_edge(START, "agent") # 流程從 START 開始，首先執行 "agent" 節點

# 條件邊：從 "agent" 節點出發，根據 trigger_queries 函式的結果決定下一步
# trigger_queries 會根據策略中的 searches 列表，為每個 search 產生一個到 "provide_answer" 的任務
agent_state.add_conditional_edges(
    "agent", # 起點節點
    trigger_queries, # 一個判斷函式，返回下一步要去哪裡或發送哪些並行任務
    ["provide_answer"] # 可能的目標節點列表 (簡化，實際更複雜)
)
# trigger_queries 函式大致如下：
async def trigger_queries(state: ThreadState, config: RunnableConfig):
    # 為策略中的每個搜尋任務，創建一個 "Send" 指令
    # 指示 LangGraph 將這些任務發送到 "provide_answer" 節點並行處理
    return [
        Send(
            "provide_answer", # 要發送到的目標節點
            { # 傳遞給該節點的 SubGraphState
                "question": state["question"],
                "instructions": s.instructions,
                "term": s.term,
            },
        )
        for s in state["strategy"].searches # 遍歷策略中的所有搜尋
    ]


# 固定邊：當 "provide_answer" 節點(的所有並行任務)完成後，流程進入 "write_final_answer" 節點
agent_state.add_edge("provide_answer", "write_final_answer")

# 固定邊：當 "write_final_answer" 節點完成後，流程結束 (END)
agent_state.add_edge("write_final_answer", END)

程式碼解釋：

4. 編譯並執行圖 (Graph)

最後，我們將定義好的圖「編譯」成一個可執行的物件，然後就可以呼叫它了。

# open_notebook/graphs/ask.py (結尾)

# 編譯圖
graph = agent_state.compile()

# 理論上，執行圖的方式如下 (實際使用可能更複雜，需要傳入設定)：
# async def run_ask_graph(user_question: str):
#     initial_state = {"question": user_question, "answers": []}
#     async for event in graph.astream(initial_state):
#         # 可以查看每一步的狀態變化
#         print(event) 
#     # 或者直接獲取最終結果
#     # final_result = await graph.ainvoke(initial_state)
#     # return final_result.get("final_answer")

# 範例：直接調用 (簡化)
# import asyncio
# async def main():
#   user_question = "我關於 AI 倫理的筆記中，提到了哪些主要觀點？"
#   # 初始狀態，注意 answers 必須是列表，因為它是 Annotated[list, operator.add]
#   initial_input = {"question": user_question, "answers": []} 
#   final_state = await graph.ainvoke(initial_input)
#   print(f"最終答案：{final_state.get('final_answer')}")

# if __name__ == "__main__":
#   asyncio.run(main())

程式碼解釋：

這就是 LangGraph 如何幫助我們將一個複雜的「智慧問答」任務，拆解成一系列定義清晰、易於管理的步驟，並自動化地執行它們。

深入探索：LangGraph 的內部運作

當我們呼叫 graph.ainvoke(...) 時，LangGraph 內部發生了什麼事呢？

  1. 初始化狀態： LangGraph 根據您提供的輸入和 ThreadState 的定義來初始化當前的工作流程狀態。
  2. 進入起點： 流程從 START 邊指向的節點開始（在我們的例子中是 agent 節點）。
  3. 執行節點： LangGraph 呼叫該節點對應的 Python 函式（例如 call_model_with_messages），並將當前的狀態傳遞給它。
  4. 更新狀態： 節點函式執行完畢後，會返回一個字典。LangGraph 用這個字典的內容來更新工作流程的狀態。例如，agent 節點返回 {"strategy": ...}，狀態中的 strategy 欄位就會被更新。
  5. 決定下一步： LangGraph 檢查從當前節點出發的邊：
    • 如果是固定邊，就直接跳到目標節點。
    • 如果是條件邊，就呼叫條件函式（例如 trigger_queries），條件函式會接收當前狀態並返回下一步的目標節點名稱（或多個 Send 指令）。
  6. 處理並行任務 (Fan-out/Fan-in)： 如果條件邊返回了多個 Send 指令（像我們的 trigger_queries 那樣），LangGraph 會為每個 Send 指令創建一個並行的任務分支。它會等待所有這些分支都執行完畢，並將它們的結果（透過 operator.add 等機制）合併回主狀態後，才會繼續執行下一個固定的邊（例如從 provide_answer 到 write_final_answer）。
  7. 重複執行： 流程跳到下一個節點，重複步驟 3-5，直到遇到 END 標記。
  8. 返回結果： 到達 END 後，LangGraph 返回最終的狀態。

以下是一個簡化的序列圖，展示了「智慧問答」流程的執行過程：

sequenceDiagram participant UserCode as 使用者程式碼 participant AskGraph as 提問圖 (LangGraph 引擎) participant StrategistNode as 策略節點 (agent) participant SearchAnswerNode as 搜尋回答節點 (provide_answer, 可能多次) participant FinalAnswerNode as 最終答案節點 (write_final_answer) UserCode->>AskGraph: ainvoke({"question": "使用者問題", "answers": []}) AskGraph->>StrategistNode: 執行 (傳入目前狀態) StrategistNode-->>AskGraph: 返回更新 {"strategy": ...} AskGraph->>AskGraph: (執行 trigger_queries 條件邊) Note over AskGraph: 根據策略，產生多個到 SearchAnswerNode 的並行任務 AskGraph->>SearchAnswerNode: 執行任務1 (傳入部分狀態) SearchAnswerNode-->>AskGraph: 返回更新 {"answers": ["片段1"]} AskGraph->>SearchAnswerNode: 執行任務2 (傳入部分狀態) SearchAnswerNode-->>AskGraph: 返回更新 {"answers": ["片段2"]} Note over AskGraph: (等待所有 SearchAnswerNode 任務完成並合併 answers) AskGraph->>FinalAnswerNode: 執行 (傳入目前狀態，包含所有 answers) FinalAnswerNode-->>AskGraph: 返回更新 {"final_answer": ...} AskGraph-->>UserCode: 返回最終狀態

open-notebook 中還有其他地方也使用了 LangGraph 來編排工作流程，例如：

這些例子都展示了 LangGraph 在定義和執行有狀態、多步驟 AI 工作流程方面的靈活性和強大功能。

總結

在本章中，我們深入了解了 LangGraph 如何作為 open-notebook 中複雜工作流程的「導演」。

LangGraph 讓開發者能夠更清晰地設計和管理複雜的 AI 應用程式邏輯。在這些由 LangGraph 編排的工作流程中，許多節點的核心任務都是與大型語言模型 (LLM) 進行互動。但是，我們如何有效地構建給這些 LLM 的指令（也就是「提示詞」）呢？如何確保提示詞既清晰又能引導模型產生我們期望的輸出格式呢？

在下一章，我們將探討 提示詞管理器 (Prompter)，了解它是如何幫助我們管理和生成這些至關重要的提示詞的。