Chapter 7: 代理人/智能體 (Agent)

歡迎來到 PocketFlow 教學的第七章！在上一章異步處理 (Async) 中，我們學習了如何讓我們的節點和流程在處理像是呼叫外部 API 或等待使用者輸入這類耗時的 I/O 操作時，不會阻塞整個應用程式的運行，使得應用程式能更流暢地回應。

然而，到目前為止，我們建立的流程 (Flow) 大多還是遵循著預先定義好的路徑。如果我們希望流程能更「聰明」一點，能像一位經驗豐富的助手那樣，根據當下的情況動態地決定下一步該做什麼呢？例如，我們提出一個問題，它能自行判斷是應該先上網搜尋資料，還是直接嘗試回答？這就是「代理人/智能體 (Agent)」概念要解決的問題。

為何需要代理人？——賦予流程思考與決策的能力

想像一下，您正在建構一個研究助理的自動化流程。您給它一個問題，例如「2024 年諾貝爾物理學獎的得主是誰？」。一個固定的流程可能只會執行單一操作，例如直接搜尋網路。但一個更智能的代理人可能會：

評估情況：我目前擁有的資訊足夠回答這個問題嗎？
選擇行動：
- 如果資訊不足，我應該「搜尋網路」。
- 如果我認為資訊已經足夠，我應該「嘗試回答」。
執行行動：執行選擇的動作。
反思與迭代：如果進行了搜尋，得到新的資訊後，我是否需要再次「搜尋網路」以獲得更多細節，還是現在可以「嘗試回答」了？

這種動態決策的能力，就是代理人的核心價值。它讓我們的流程不再是死板的指令序列，而是能根據情境做出判斷的智能實體。

代理人/智能體 (Agent) 是一種設計模式，賦予流程動態決策的能力。就像一位經驗豐富的專案經理，能根據當前情境（上下文資訊）、歷史行動和可用工具（行動空間），自主判斷並選擇下一步最合適的行動。這通常透過一個特殊的節點 (Node)（我們稱之為「代理人節點」）實現，該節點利用大型語言模型（LLM）分析上下文並決定流程的走向，可以實現如搜索、回答、或其他自定義工具調用等複雜行為。

圖示：代理人根據情境選擇下一步行動

代理人的核心組成

在 PocketFlow 中，一個代理人通常由以下幾個關鍵部分組成：

代理人節點 (Agent Node)：這是代理人的「大腦」。它通常是一個特殊的節點 (Node)，其 exec 方法會：
- 收集來自共享儲存 (Shared Store) 的上下文資訊（例如：原始任務、先前的行動、已有的結果）。
- 將這些資訊以及可用的「行動空間」（即代理人可以選擇執行的工具或動作列表）組織成一個提示 (prompt)。
- 呼叫大型語言模型 (LLM)，讓 LLM 根據提示來決定下一步應該執行哪個行動，以及執行該行動所需的參數。
- 解析 LLM 的回應，並將選擇的行動名稱作為其 post 方法的回傳值，以指導流程 (Flow) 的走向。
工具節點 (Tool Nodes)：這些是執行具體任務的普通節點 (Node)。例如，一個 SearchWebNode（搜尋網路的節點）、一個 CalculateNode（執行計算的節點）或一個 AnswerQuestionNode（產生答案的節點）。代理人節點會決定呼叫哪個工具節點。
流程 (Flow)：將代理人節點和工具節點連接起來。流程的設計通常包含：
- 從代理人節點出發，根據其回傳的行動指令，分支到相應的工具節點。
- 工具節點執行完畢後，通常會將流程導回到代理人節點，讓代理人可以根據新的情境（例如工具執行的結果）決定下一步。這形成了一個「思考-行動-觀察」的循環。
共享儲存 (Shared Store)：用於在代理人節點和工具節點之間傳遞數據，包括：
- 任務的初始描述。
- 代理人做出的歷史決策和行動。
- 工具節點執行的結果。
- 任何其他相關的上下文資訊。

小試身手：打造一個簡易研究代理人

讓我們來實作一個簡易的研究代理人。它會接收一個問題，然後決定是直接回答，還是先進行網路搜尋。

步驟 1：定義工具節點

我們需要兩個工具節點：一個用於搜尋，一個用於回答。

SearchWebNode (搜尋網路節點)

from pocketflow import Node
import yaml # 用於解析 LLM 的結構化輸出

# 模擬的網路搜尋函數
def search_web_mock(query: str) -> str:
    print(f"🔎 模擬搜尋：{query}")
    if "諾貝爾物理學獎" in query:
        return "研究顯示，2024年諾貝爾物理學獎的資訊尚未廣泛公布。"
    return "找不到相關資訊。"

class SearchWebNode(Node):
    def prep(self, shared):
        # 從共享儲存獲取代理人決定的搜尋詞
        return shared.get("current_search_term", "未知搜尋詞")

    def exec(self, search_term: str):
        return search_web_mock(search_term)

    def post(self, shared, prep_res, exec_res: str):
        # 將搜尋結果添加到共享儲存的上下文中
        history = shared.get("history", [])
        history.append({"action": "search", "term": prep_res, "result": exec_res})
        shared["history"] = history
        print(f"📝 搜尋結果已記錄：{exec_res[:30]}...")
        return "action_completed" # 行動完成，返回給代理人決策

prep: 從共享儲存 (Shared Store) 取得由代理人節點決定的 current_search_term。
exec: 執行模擬的網路搜尋。
post: 將搜尋的行動和結果記錄到共享儲存 (Shared Store) 的 history 中，並回傳 "action_completed"，表示此工具已執行完畢，流程應回到代理人節點進行下一步決策。

AnswerQuestionNode (回答問題節點)

# 模擬的 LLM 回答函數
def llm_answer_mock(query: str, context: str) -> str:
    print(f"💬 模擬 LLM 根據上下文回答：'{query}' (上下文：'{context[:30]}...')")
    if "諾貝爾物理學獎" in query and "尚未廣泛公布" in context:
        return "根據目前搜尋到的資訊，2024年諾貝爾物理學獎的得主資訊尚未廣泛公布。"
    return "我需要更多資訊才能回答這個問題。"

class AnswerQuestionNode(Node):
    def prep(self, shared):
        query = shared.get("original_query", "未知問題")
        history_str = str(shared.get("history", [])) # 將歷史轉換為字串
        return query, history_str

    def exec(self, inputs: tuple):
        query, history_context = inputs
        return llm_answer_mock(query, history_context)

    def post(self, shared, prep_res, exec_res: str):
        shared["final_answer"] = exec_res
        print(f"✅ 最終答案：{exec_res}")
        # 預設回傳 "default"，通常表示代理任務的一個分支結束

prep: 從共享儲存 (Shared Store) 獲取原始問題和所有歷史記錄作為上下文。
exec: 呼叫模擬的 LLM 來根據上下文產生答案。
post: 將最終答案儲存到共享儲存 (Shared Store)。

步驟 2：定義代理人節點 (Agent Node)

這是代理人的核心，我們稱之為 DecideActionNode。

# 模擬的 LLM 決策函數
def llm_decide_action_mock(prompt: str) -> str:
    print("🤔 代理人正在思考...")
    print(f"提示 (前100字)：{prompt[:100]}...")
    # 簡化邏輯：如果歷史記錄為空，則搜尋；否則回答
    if "Previous Actions: []" in prompt or "Previous Actions: No previous actions" in prompt :
        response_yaml = """
action: search
parameters:
  search_term: "2024年諾貝爾物理學獎得主"
"""
    else:
        response_yaml = """
action: answer
parameters: {}
"""
    print(f"代理人 LLM 回應 (模擬)：\n{response_yaml}")
    return response_yaml

class DecideActionNode(Node):
    def prep(self, shared):
        query = shared.get("original_query", "沒有提供問題")
        history = shared.get("history", []) # 先前行動的歷史
        # 建立上下文資訊
        context_info = f"原始問題：{query}\n先前行動：{history if history else '無先前行動'}"
        return context_info

    def exec(self, context: str):
        # 建立給 LLM 的提示
        prompt = f"""
### 情境
{context}

### 可用行動
1. search: 進行網路搜尋以獲取資訊。
   參數:
     - search_term (str): 要搜尋的關鍵字。
2. answer: 根據目前擁有的資訊回答問題。
   參數: (無)

### 下一步行動
根據當前情境和可用行動，決定下一步。
請以 YAML 格式回傳您的思考過程和決定：
```yaml
thinking: |
  (您的思考步驟)
action: <行動名稱>
parameters:
  <參數名稱>: <參數值>

""" llm_response_yaml = llm_decide_action_mock(prompt) # 解析 LLM 的 YAML 回應 # 在真實應用中，這裡需要更穩健的 YAML 解析和錯誤處理 try: parsed_response = yaml.safe_load(llm_response_yaml) if not isinstance(parsed_response, dict) or "action" not in parsed_response: # 如果解析失敗或格式不對，預設為一個安全動作，例如再次嘗試或結束 print("⚠️ LLM 回應格式錯誤，將嘗試結束。") return {"action": "error_fallback", "thinking": "LLM 回應格式錯誤"} except yaml.YAMLError: print("⚠️ LLM 回應 YAML 解析失敗，將嘗試結束。") return {"action": "error_fallback", "thinking": "LLM 回應 YAML 解析失敗"}

    return parsed_response # 例如：{"action": "search", "parameters": {"search_term": "..."}}

def post(self, shared, prep_res, exec_res: dict):
    chosen_action = exec_res.get("action", "error_fallback")
    parameters = exec_res.get("parameters", {})

    print(f"代理人決定行動：{chosen_action}，參數：{parameters}")

    if chosen_action == "search":
        shared["current_search_term"] = parameters.get("search_term")
    # 其他行動的參數也可以在這裡設定到 shared store
    
    # 將 LLM 的思考過程也記錄下來，方便除錯
    shared.setdefault("agent_thoughts", []).append(exec_res.get("thinking", ""))

    return chosen_action # 回傳 LLM 決定的行動名稱

*   `prep`: 從[共享儲存 (Shared Store)](02_共享儲存__shared_store__.md) 收集原始問題和歷史行動。
*   `exec`:
    1.  建構一個詳細的提示，包含當前情境和可用的行動空間（`search` 和 `answer`）。
    2.  呼叫（模擬的）LLM `llm_decide_action_mock`，它會回傳一個 YAML 格式的字串，指明選擇的行動和所需參數。
    3.  解析 YAML 回應。
*   `post`:
    1.  從解析後的 LLM 回應中提取行動名稱和參數。
    2.  如果行動是 `search`，則將 `search_term` 存入[共享儲存 (Shared Store)](02_共享儲存__shared_store__.md)，供 `SearchWebNode` 使用。
    3.  回傳 LLM 選擇的行動名稱（例如 `"search"` 或 `"answer"`）。這個回傳值將被[流程 (Flow)](03_流程__flow__.md) 用來決定下一個執行的節點。

### 步驟 3：建立流程 (Flow) 並連接節點

現在我們將這些節點連接起來，形成一個可以循環決策的流程。

```python
from pocketflow import Flow

# 建立節點實例
decide_node = DecideActionNode()
search_node = SearchWebNode()
answer_node = AnswerQuestionNode()

# 定義轉換規則
# 代理人決定搜尋 -> 執行搜尋節點
decide_node - "search" >> search_node
# 代理人決定回答 -> 執行回答節點
decide_node - "answer" >> answer_node
# 代理人發生錯誤或無法決策 -> 結束 (這裡我們可以連到一個結束節點或不連)
# decide_node - "error_fallback" >> some_end_node 

# 搜尋節點完成後 -> 回到代理人節點進行下一步決策
search_node - "action_completed" >> decide_node

# 建立流程，以代理人節點為起始點
research_agent_flow = Flow(start=decide_node)

這個流程的關鍵在於 search_node - "action_completed" >> decide_node，它使得在執行完搜尋後，流程會回到 decide_node，讓代理人可以根據新的搜尋結果再次決策。

步驟 4：運行代理人流程

# 準備共享儲存，放入初始問題
shared_data = {"original_query": "2024年諾貝爾物理學獎得主是誰？"}

print(f"🚀 開始研究代理人流程，問題：{shared_data['original_query']}")
research_agent_flow.run(shared_data)
print("✨ 研究代理人流程結束")

if "final_answer" in shared_data:
    print(f"\n💡 代理人的最終答案：{shared_data['final_answer']}")
print(f"📜 代理人歷史行動：{shared_data.get('history')}")

當您運行此程式碼時，您將看到代理人首先決定搜尋，然後執行搜尋，接著根據搜尋結果（在此模擬中，結果表明資訊不足），再次決策並選擇回答。

預期輸出範例（簡化）：

🚀 開始研究代理人流程，問題：2024年諾貝爾物理學獎得主是誰？
🤔 代理人正在思考...
提示 (前100字)：
### 情境
原始問題：2024年諾貝爾物理學獎得主是誰？
先前行動：無先前行動

### 可用行動
1. search: 進行網路搜尋以獲取資訊。
...
代理人 LLM 回應 (模擬)：
action: search
parameters:
  search_term: "2024年諾貝爾物理學獎得主"
代理人決定行動：search，參數：{'search_term': '2024年諾貝爾物理學獎得主'}
🔎 模擬搜尋：2024年諾貝爾物理學獎得主
📝 搜尋結果已記錄：研究顯示，2024年諾貝爾物理學獎的資訊尚未廣泛公...
🤔 代理人正在思考...
提示 (前100字)：
### 情境
原始問題：2024年諾貝爾物理學獎得主是誰？
先前行動：[{'action': 'search', 'term': '2024年諾貝爾物理學獎得主', 'result': '研究顯示，2024年諾貝爾物理學獎的資訊尚未廣泛公布。'}]

...
代理人 LLM 回應 (模擬)：
action: answer
parameters: {}
代理人決定行動：answer，參數：{}
💬 模擬 LLM 根據上下文回答：'2024年諾貝爾物理學獎得主是誰？' (上下文："{'action': 'search', 'term': '...")
✅ 最終答案：根據目前搜尋到的資訊，2024年諾貝爾物理學獎的得主資訊尚未廣泛公布。
✨ 研究代理人流程結束

💡 代理人的最終答案：根據目前搜尋到的資訊，2024年諾貝爾物理學獎的得主資訊尚未廣泛公布。
📜 代理人歷史行動：[{'action': 'search', 'term': '2024年諾貝爾物理學獎得主', 'result': '研究顯示，2024年諾貝爾物理學獎的資訊尚未廣泛公布。'}]

這個簡單的例子展示了代理人如何透過 LLM 進行動態決策，並在不同的工具節點間導航。

設計高效能、高可靠度代理人的關鍵

要建立出色的代理人，以下幾點至關重要：

上下文管理 (Context Management)：
- 提供相關且最精簡的上下文給 LLM。過多的無關資訊可能會干擾 LLM 的判斷。
- 例如，與其將整個聊天歷史都丟給 LLM，不如使用像我們將在下一章討論的檢索增強生成 (RAG) 技術來提取最相關的片段。
- 即使 LLM 的上下文視窗越來越大，它們仍然可能受到「中間內容遺失 (lost in the middle)」問題的影響，即忽略提示中間部分的資訊。
行動空間設計 (Action Space Design)：
- 提供一個結構清晰、無歧義的行動（工具）集合。
- 避免行動功能重疊，例如同時提供 read_databases 和 read_csvs 兩個行動。更好的做法可能是將 CSV 匯入資料庫，然後只提供一個統一的資料庫讀取行動。

優良行動設計範例

增量式處理 (Incremental)：將內容分塊（例如一次處理 500 行或一頁）餵給代理人或工具，而不是一次處理全部內容。
概覽後深入 (Overview-zoom-in)：先提供高層次的結構（例如目錄、摘要），然後允許代理人深入查看細節（例如原始文本）。
參數化/可程式化 (Parameterized/Programmable)：與其提供固定的行動，不如讓行動可以接受參數（例如：選擇哪些欄位）或甚至可程式化（例如：執行 SQL 查詢），以增加靈活性。
回溯 (Backtracking)：允許代理人撤銷上一步的行動，而不是在遇到錯誤或死胡同時完全重新開始，這樣可以保留已有的進展。

代理人內部運作機制

代理人在 PocketFlow 中的實現，並非依賴一個名為 Agent 的特殊類別，而是一種設計模式，巧妙地結合了我們已經學過的節點 (Node)、流程 (Flow) 和共享儲存 (Shared Store)。

非程式碼逐步解析：

啟動：流程 (Flow) 開始執行，通常第一個節點就是代理人節點（例如我們的 DecideActionNode）。
代理人節點 - prep：從共享儲存 (Shared Store) 收集當前的任務描述、歷史行動、以及任何由先前工具節點產生的結果。
代理人節點 - exec：
- 將收集到的上下文資訊，連同定義好的可用行動列表（行動空間），組合成一個提示 (prompt)。
- 呼叫大型語言模型 (LLM)，傳遞這個提示。
- LLM 分析提示後，回傳它選擇的下一個行動以及執行該行動所需的參數（通常是結構化格式，如 YAML 或 JSON）。
- exec 方法解析 LLM 的回應。
代理人節點 - post：
- 將 LLM 回應中包含的參數（例如，如果選擇的行動是「搜尋」，則參數可能是「搜尋關鍵字」）存入共享儲存 (Shared Store)，以供後續的工具節點使用。
- 回傳 LLM 選擇的行動名稱（例如 "search" 或 "answer"）作為此節點的行動指令。
流程導航：流程 (Flow) 根據代理人節點回傳的行動指令，以及預先定義的轉換規則（例如 decide_node - "search" >> search_node），決定下一個要執行的工具節點。
工具節點執行：被選中的工具節點執行其任務（例如，SearchWebNode 進行網路搜尋）。它會從共享儲存 (Shared Store) 讀取所需的參數，並將其執行結果寫回共享儲存 (Shared Store)。
循環 (Loop)：工具節點執行完畢後，其 post 方法通常會回傳一個特定的行動指令，該指令會將流程 (Flow) 導回到代理人節點。
代理人節點再次從步驟 2 開始，根據更新後的上下文（包含了剛才工具節點的執行結果）進行新一輪的決策。這個循環會一直持續，直到代理人決定執行一個終止行動（例如「回答」且不再需要更多資訊），或者達到預設的結束條件。

序列圖 (Sequence Diagram) 概覽：

sequenceDiagram participant 使用者 participant 研究代理流程 as AgentFlow participant 決策節點 as DecideNode (代理人) participant 搜尋節點 as SearchNode (工具) participant LLM服務 as LLMService participant 共享儲存 as SharedStore 使用者->>研究代理流程: run({"original_query": "問題"}) 研究代理流程->>決策節點: _run(shared_data) 決策節點->>共享儲存: 讀取 original_query, history 共享儲存-->>決策節點: (查詢和歷史記錄) 決策節點->>LLM服務: (建構提示) 分析上下文，選擇行動 (搜尋/回答) LLM服務-->>決策節點: 回應：行動："search", 參數：{"search_term": "關鍵字"} 決策節點->>共享儲存: 儲存 "current_search_term" = "關鍵字" 決策節點->>研究代理流程: 回傳行動 "search" 研究代理流程->>搜尋節點: _run(shared_data) 搜尋節點->>共享儲存: 讀取 "current_search_term" 搜尋節點->>搜尋節點: (執行 search_web_mock) 搜尋節點->>共享儲存: 更新 history (加入搜尋結果) 搜尋節點->>研究代理流程: 回傳行動 "action_completed" 研究代理流程->>決策節點: _run(shared_data) (再次決策) %% ... 流程可能繼續循環，或最終導向 AnswerNode ...%% end

這個序列圖展示了代理人流程如何協調決策節點、工具節點、LLM 和共享儲存，以完成一個動態的任務。

總結

在本章中，我們深入探索了 PocketFlow 中的「代理人/智能體 (Agent)」設計模式：

代理人賦予流程動態決策的能力，使其能根據上下文、歷史和可用工具自主選擇下一步行動。
代理人的核心通常是一個代理人節點，它利用 LLM 進行思考和決策。
代理人與一組工具節點協同工作，透過精心設計的流程進行連接和循環。
共享儲存在代理人和工具之間傳遞上下文資訊和結果。
我們透過一個「簡易研究代理人」的範例，學習了如何建構和運行一個基本的代理人流程。
高效的上下文管理和清晰的行動空間設計是建立強大代理人的關鍵。

代理人模式開啟了自動化流程的全新可能性，讓它們能夠處理更複雜、更需要「智能」的任務。

到目前為止，我們的代理人依賴於即時收集的上下文。但如果我們希望代理人能夠利用一個龐大且預先準備好的知識庫來增強其決策和回答能力呢？這就引出了我們下一章，也是本教學系列最後一章的主題：檢索增強生成 (RAG)。