總字數：約22000字 | 預計閱讀時長：60-70分鐘​

一、LangGraph概述與定位​

1.1 什麼是LangGraph​

LangGraph 是LangChain團隊於2024年推出的開源框架，專門用於構建有狀態的、多智慧體的LLM應用。它基於圖計算模型，讓開發者能夠以節點（Node）和邊（Edge）的方式編排複雜的AI工作流。​

如果說LangChain是"AI樂高積木"，那麼LangGraph就是"AI流程圖編輯器"——它讓你能夠直觀地定義AI應用的執行路徑，支援迴圈、分支、並行等複雜控制流。​

LangGraph的核心特性：​

​



| 特性 | 說明 | 價值 |
| --- | --- | --- |
| 圖結構 | 基於有向圖的執行模型 | 支援複雜的控制流 |
| 狀態管理 | 內建狀態定義與傳遞機制 | 支援有狀態的多輪互動 |
| 檢查點 | 自動儲存執行狀態 | 支援中斷恢復與除錯 |
| 人機協作 | 內建Human-in-the-Loop | 支援人工稽核與幹預 |
| 流式輸出 | 節點級流式處理 | 實時反饋執行進度 |
| 多智慧體 | 原生支援多Agent協作 | 構建複雜的智慧體團隊 |



​

1.2 為什麼需要LangGraph​

1）LangChain的侷限性​

LangChain的鏈式呼叫模型雖然簡潔，但在構建複雜應用時存在明顯瓶頸：​

線性執行的限制：​

•

Chains是線性執行的，A → B → C，難以實現迴圈和分支​

•

複雜的業務邏輯需要多層巢狀，程式碼可讀性差​

•

狀態管理依賴外部元件，不夠直觀​

多智慧體協作的挑戰：​

•

缺乏原生的多Agent編排能力​

•

Agent之間的通訊需要手動實現​

•

難以實現複雜的協作模式（如主從、對等、層級）​

實際開發中的痛點：​

​

Code block​

Python

\# 傳統LangChain實現ReAct迴圈的複雜程式碼​

from langchain.chains import LLMChain​

from langchain.memory import ConversationBufferMemory​

​

\# 需要手動管理迴圈和狀態​

memory = ConversationBufferMemory()​

​

2）圖結構的優勢​

LangGraph用圖結構解決了這些問題：​

直觀的控制流：​

•

節點代表計算單元，邊代表執行路徑​

•

條件邊支援動態路由，迴圈邊支援迭代​

•

複雜邏輯視覺化，易於理解和維護​

原生狀態管理：​

•

狀態在節點間自動傳遞​

•

支援狀態的部分更新​

•

檢查點機制支援斷點續傳​

LangChain vs LangGraph對比：​

​



| 維度 | LangChain | LangGraph |
| --- | --- | --- |
| 執行模型 | 線性鏈式 | 圖結構 |
| 控制流 | 順序執行 | 迴圈、分支、並行 |
| 狀態管理 | 外部元件 | 內建狀態機 |
| 多智慧體 | 需手動實現 | 原生支援 |
| 視覺化 | 不支援 | 支援圖視覺化 |
| 除錯能力 | 有限 | 檢查點 + 時間旅行 |



​

3）有狀態應用的需求場景​

為什麼狀態管理如此重要？​

在真實的AI應用中，大多數場景都需要維護狀態：​

​



| 場景 | 狀態需求 | 傳統方案的痛點 |
| --- | --- | --- |
| 多輪對話 | 對話歷史、使用者偏好 | 手動管理，容易丟失 |
| 任務執行 | 執行進度、中間結果 | 無法中斷恢復 |
| 多智慧體協作 | 共享知識、任務分配 | 通訊機制複雜 |
| 人工稽核 | 稽核狀態、反饋記錄 | 流程難以串聯 |



​

1.3 LangGraph在Agent開發中的地位​

1）技術棧演進​

AI應用開發框架的演進路線：​

​

Code block​

Plain Text

​

LangGraph在技術棧中的位置：​

​

Code block​

Plain Text

​

2）適用場景對比​

何時選擇LangChain，何時選擇LangGraph？​

​



| 應用型別 | 推薦框架 | 理由 |
| --- | --- | --- |
| 簡單的單輪對話 | LangChain | 輕量級，開箱即用 |
| 線性RAG流水線 | LangChain | Chains足夠應對 |
| ReAct Agent | LangGraph | 需要迴圈和狀態 |
| 多智慧體協作 | LangGraph | 原生支援複雜編排 |
| 人機協作流程 | LangGraph | 內建Human-in-the-Loop |
| 長任務執行 | LangGraph | 檢查點支援中斷恢復 |



​

安全場景中的典型應用：​

•

漏洞掃描流水線：多階段掃描，需要狀態追蹤和錯誤恢復​

•

威脅情報分析：多源資料融合，需要並行處理和結果聚合​

•

自動化應急響應：多角色協作，需要人工稽核和決策點​

•

安全報告生成：多步驟生成，需要狀態持久化和斷點續傳​

二、LangGraph核心概念​

2.1 圖（Graph）​

1）有向圖的基本結構​

圖（Graph） 是LangGraph的核心抽象，它由\*\*節點（Node）和邊（Edge）\*\*組成。​

圖的基本組成：​

​

Code block​

Plain Text

​

圖的型別：​

​



| 圖型別 | 特點 | 適用場景 |
| --- | --- | --- |
| StateGraph | 帶狀態的圖，最常用 | 大多數應用場景 |
| MessageGraph | 以訊息列表為狀態 | 簡單的對話場景 |



​

2）節點（Node）與邊（Edge）​

節點是圖中的計算單元，每個節點是一個Python函式，接收狀態作為輸入，返回更新後的狀態。​

邊定義了節點之間的連線關係，決定了執行流程。​

基本圖結構示例：​

​

Code block​

Python

\# basic\_graph\_structure.py​

from langgraph.graph import StateGraph, END​

from typing import TypedDict, Annotated​

import operator​

​

\# 定義狀態結構​

class GraphState(TypedDict):​

messages: Annotated\[list, operator.add\] \# 訊息列表，使用add運算子合並​

current\_step: str \# 當前步驟​

result: str \# 處理結果​

​

\# 建立狀態圖​

​

3）條件邊與分支邏輯​

條件邊允許根據狀態動態選擇執行路徑，實現分支邏輯。​

​

Code block​

Python

\# conditional\_edge.py​

from langgraph.graph import StateGraph, END​

from typing import TypedDict, Literal​

​

class AnalysisState(TypedDict):​

vulnerability\_type: str​

severity: str​

needs\_manual\_review: bool​

​

​

2.2 狀態（State）​

1）狀態的定義與管理​

狀態（State） 是LangGraph的核心概念，它定義了在圖執行過程中需要維護的所有資料。​

狀態定義的最佳實踐：​

​

Code block​

Python

\# state\_definition.py​

from typing import TypedDict, Annotated, List, Optional, Dict, Any​

from langgraph.graph import add\_messages​

from langchain\_core.messages import BaseMessage​

​

class SecurityAnalysisState(TypedDict):​

"""安全分析狀態定義"""​

​

\# 訊息歷史，使用add\_messages運算子自動追加​

messages: Annotated\[List\[BaseMessage\], add\_messages\]​

​

\# 當前分析階段​

stage: str \# "scanning" | "analyzing" | "reporting"​

​

\# 掃描結果​

scan\_results: List\[Dict\[str, Any\]\]​

​

\# 發現的漏洞​

vulnerabilities: List\[Dict\[str, Any\]\]​

​

\# 分析結論​

analysis\_report: Optional\[str\]​

​

\# 錯誤資訊​

errors: List\[str\]​

​

\# 配置引數​

​

狀態運算子：​

​



| 運算子 | 說明 | 使用場景 |
| --- | --- | --- |
| add | 追加到列表 | 訊息歷史、日誌 |
| replace | 替換值 | 狀態更新、結果覆蓋 |
| update | 合併字典 | 配置更新 |



​

2）狀態的傳遞與更新​

狀態在節點間的傳遞機制：​

1.

圖啟動時，傳入初始狀態​

2.

每個節點接收當前狀態作為輸入​

3.

節點返回狀態的部分更新（不是完整狀態）​

4.

LangGraph自動合併更新到狀態中​

5.

更新後的狀態傳遞給下一個節點​

​

Code block​

Python

\# state\_update.py​

from typing import Dict, Any​

​

def scanner\_node(state: SecurityAnalysisState) -> Dict\[str, Any\]:​

"""掃描節點：只返回需要更新的部分"""​

\# 讀取狀態​

target = state.get("config", {}).get("target")​

​

3）檢查點（Checkpoint）機制​

檢查點是LangGraph的強大特性，它允許在圖執行的每個步驟自動儲存狀態快照。​

檢查點的價值：​

​



| 功能 | 說明 | 應用場景 |
| --- | --- | --- |
| 中斷恢復 | 從任意節點恢復執行 | 長任務中斷後繼續 |
| 時間旅行 | 回溯到歷史狀態 | 除錯和問題排查 |
| 人工幹預 | 在特定節點暫停等待人工輸入 | 稽核流程 |
| 狀態持久化 | 將狀態儲存到資料庫 | 生產環境部署 |



​

​

Code block​

Python

\# checkpoint\_demo.py​

from langgraph.graph import StateGraph​

from langgraph.checkpoint.memory import MemorySaver​

​

​

5.3 人機協作（Human-in-the-Loop）​

1）人工稽核節點的設計​

Human-in-the-Loop是LangGraph的核心特性，讓AI應用能夠在關鍵節點暫停，等待人工確認或輸入。​

​

Code block​

Python

\# human\_in\_loop.py​

from typing import TypedDict, Annotated, List, Optional​

from langgraph.graph import StateGraph, END​

from langgraph.checkpoint.memory import MemorySaver​

from langchain\_core.messages import BaseMessage​

from langgraph.graph import add\_messages​

​

class HITLState(TypedDict):​

messages: Annotated\[List\[BaseMessage\], add\_messages\]​

analysis\_result: Optional\[str\]​

human\_feedback: Optional\[str\]​

approved: bool​

final\_output: Optional\[str\]​

​

def analysis\_node(state: HITLState) -> dict:​

"""自動分析節點"""​

return {​

"analysis\_result": "發現3個高危漏洞",​

"messages": \[("system", "自動分析完成，等待人工確認")\]​

}​

​

def human\_approval\_node(state: HITLState) -> dict:​

"""人工審批節點（在此處中斷）"""​

\# 此函式在中斷恢復後執行​

feedback = state.get("human\_feedback", "")​

approved = "同意" in feedback or "透過" in feedback or approved​

​

return {​

​

2）中斷與繼續機制​

中斷型別：​

​



| 型別 | 說明 | 使用場景 |
| --- | --- | --- |
| interrupt\_before | 在節點執行前中斷 | 人工確認後執行 |
| interrupt\_after | 在節點執行後中斷 | 審查執行結果 |



​

3）安全場景中的應用​

在安全場景中，Human-in-the-Loop非常關鍵：​

​



| 場景 | 中斷點 | 人工操作 |
| --- | --- | --- |
| 漏洞修復 | 修復執行前 | 確認修復方案 |
| 應急響應 | 措施執行前 | 審批響應措施 |
| 報告發布 | 報告傳送前 | 稽核報告內容 |
| 許可權變更 | 變更執行前 | 確認變更請求 |



​

5.4 錯誤處理與重試​

1）異常捕獲機制​

​

Code block​

Python

\# error\_handling.py​

from typing import TypedDict, Annotated, List​

from langgraph.graph import StateGraph, END​

from langchain\_core.messages import BaseMessage​

from langgraph.graph import add\_messages​

​

class RobustState(TypedDict):​

messages: Annotated\[List\[BaseMessage\], add\_messages\]​

retry\_count: int​

max\_retries: int​

error: str​

result: str​

​

def risky\_node(state: RobustState) -> dict:​

"""可能失敗的節點"""​

import random​

​

try:​

\# 模擬可能失敗的操作​

if random.random() < 0.5:​

​

2）自動重試策略​

​

Code block​

Python

\# retry\_with\_backoff.py​

import time​

from typing import TypedDict, Annotated, List​

​

class RetryState(TypedDict):​

​

六、安全實戰案例​

6.1 智慧漏洞分析系統​

1）需求分析與架構設計​

系統目標：構建一個能夠自動掃描、分析和報告漏洞的智慧系統。​

架構設計：​

​

Code block​

Plain Text

使用者輸入目標​

│​

▼​

┌─────────────────────────────────────┐​

│ Supervisor Agent │​

│ (任務規劃與排程) │​

└───────┬─────────┬─────────┬─────────┘​

│ │ │​

▼ ▼ ▼​

┌────────┐ ┌────────┐ ┌────────┐​

│Scanner │ │Analyzer│ │Reporter│​

│ Agent │ │ Agent │ │ Agent │​

└────────┘ └────────┘ └────────┘​

│ │ │​

└─────────┴─────────┘​

│​

▼​

最終安全報告​

​

2）多智慧體協作流程​

3）完整程式碼實現​

​

Code block​

Python

\# vuln\_analysis\_system.py​

from typing import TypedDict, Annotated, List, Dict, Any, Optional​

from langgraph.graph import StateGraph, END​

from langgraph.checkpoint.memory import MemorySaver​

from langchain\_core.messages import BaseMessage, HumanMessage, SystemMessage​

from langgraph.graph import add\_messages​

from langchain\_deepseek import ChatDeepSeek​

from langchain\_core.tools import tool​

from langgraph.prebuilt import ToolNode​

import os​

from dotenv import load\_dotenv​

​

load\_dotenv()​

​

class VulnAnalysisState(TypedDict):​

messages: Annotated\[List\[BaseMessage\], add\_messages\]​

target: str​

phase: str \# "init" | "scanning" | "analyzing" | "reporting" | "done"​

scan\_results: Dict\[str, Any\]​

vulnerabilities: List\[Dict\[str, Any\]\]​

risk\_level: str \# "low" | "medium" | "high" | "critical"​

report: Optional\[str\]​

next\_agent: str​

​

llm = ChatDeepSeek(​

model="deepseek-chat",​

api\_key=os.getenv("DEEPSEEK\_API\_KEY"),​

temperature=0.3​

)​

​

​

6.2 自動化安全報告生成​

1）資料收集智慧體​

​

Code block​

Python

\# data\_collector\_agent.py​

from typing import TypedDict, Dict, Any​

​

2）分析研判智慧體​

​

Code block​

Python

\# analysis\_agent.py​

​

3）報告撰寫智慧體​

​

Code block​

Python

\# report\_writer\_agent.py​

def report\_writing\_node(state: dict) -> dict:​

"""報告撰寫節點：生成結構化的安全報告"""​

analysis = state.get("analysis\_result", {})​

collected\_data = state.get("collected\_data", {})​

​

report = f"""​

​

6.3 威脅情報處理流水線​

​

Code block​

Python

\# threat\_intel\_pipeline.py​

from typing import TypedDict, Annotated, List, Dict​

from langgraph.graph import StateGraph, END​

from langchain\_core.messages import BaseMessage​

from langgraph.graph import add\_messages​

​

class ThreatIntelState(TypedDict):​

messages: Annotated\[List\[BaseMessage\], add\_messages\]​

ioc\_list: List\[str\] \# 失陷指標列表​

enrichment\_data: Dict​

​

七、LangGraph vs 其他框架​

7.1 框架對比​

​



| 維度 | LangGraph | AutoGen | CrewAI | MetaGPT |
| --- | --- | --- | --- | --- |
| 開發公司 | LangChain | Microsoft | CrewAI | DeepWisdom |
| 核心理念 | 圖結構編排 | 對話式協作 | 角色扮演 | 軟體公司模擬 |
| 狀態管理 | 內建檢查點 | 會話歷史 | 共享上下文 | 全域性狀態 |
| 多智慧體 | 原生支援 | 核心特性 | 核心特性 | 核心特性 |
| 人機協作 | 原生支援 | 支援 | 有限 | 有限 |
| 視覺化 | LangGraph Studio | AutoGen Studio | 無 | 無 |
| 學習曲線 | 中等 | 中等 | 較低 | 較高 |
| 社群活躍度 | 高 | 高 | 中等 | 中等 |
| 生產就緒 | 是 | 是 | 發展中 | 發展中 |



​

7.2 選型建議​

選擇LangGraph的理由：​

​



| 場景 | 推薦理由 |
| --- | --- |
| 需要複雜控制流 | 圖結構天然支援迴圈、分支、並行 |
| 需要狀態持久化 | 內建檢查點機制，支援中斷恢復 |
| 需要人機協作 | 原生Human-in-the-Loop支援 |
| 已使用LangChain生態 | 無縫整合，複用現有元件 |
| 需要生產級穩定性 | LangChain團隊維護，持續更新 |



​

選擇AutoGen的理由：​

​



| 場景 | 推薦理由 |
| --- | --- |
| 對話式協作 | Agent之間透過對話協作 |
| 程式碼執行 | 內建程式碼執行沙箱 |
| 微軟生態 | 與Azure深度整合 |



​

選擇CrewAI的理由：​

​



| 場景 | 推薦理由 |
| --- | --- |
| 角色扮演場景 | 自然的角色定義方式 |
| 快速原型 | 學習曲線低，快速上手 |
| 簡單協作 | 不需要複雜的控制流 |



​

八、總結與展望​

8.1 核心要點回顧​

LangGraph的核心價值：​

1.

圖結構編排：用節點和邊定義複雜的執行流程，支援迴圈、分支、並行​

2.

狀態管理：內建狀態機，支援狀態持久化和斷點恢復​

3.

多智慧體協作：原生支援Supervisor、Peer-to-Peer等多種協作模式​

4.

人機協作：Human-in-the-Loop讓AI應用更加可控​

關鍵技術點：​

​



| 概念 | 說明 | 應用 |
| --- | --- | --- |
| StateGraph | 帶狀態的圖 | 大多數應用場景 |
| 節點（Node） | 計算單元 | 執行具體邏輯 |
| 邊（Edge） | 連線關係 | 定義執行流程 |
| 條件邊 | 動態路由 | 實現分支邏輯 |
| 檢查點 | 狀態快照 | 中斷恢復、除錯 |
| ToolNode | 工具節點 | 整合外部工具 |



​

學習路徑建議：​

​

Code block​

Plain Text

基礎：LangChain Chains → 理解鏈式呼叫​

​

8.2 未來發展趨勢​

多模態智慧體：​

•

結合視覺、語音、文字的多模態理解​

•

安全場景：影象識別漏洞、語音指令執行​

自主協作能力：​

•

Agent能夠自動發現和組建團隊​

•

動態調整協作策略​

安全領域的應用前景：​

​



| 方向 | 應用 |
| --- | --- |
| 自動化滲透測試 | 多Agent協作完成完整滲透流程 |
| 智慧應急響應 | 自動化事件檢測、分析、處置 |
| 安全運營中心 | AI驅動的SOC，自動告警研判和處置 |
| 威脅狩獵 | 主動搜尋和發現高階威脅 |



​

​

​

總結：LangGraph作為LangChain生態的重要補充，為構建複雜的、有狀態的AI應用提供了強大的框架支援。在安全領域，它的圖結構編排、狀態管理和多智慧體協作能力，讓我們能夠構建更加智慧、可靠的安全工具和平臺。掌握LangGraph，將為你的安全AI開發生涯增添一項核心競爭力。​

​

​

作者：無涯": {"thread\_id": "analysis-001"}}​

result = app.invoke(initial\_state, config=config)​

後續可以從檢查點恢復​

state = app.get\_state(config)​

print(f"當前狀態：{state.values}")​

print(f"下一步：{state.next}")​

​

Code block​

Plain Text

​

\## 2.3 節點（Node）​

​

\### 1）節點的型別與職責​

​

LangGraph中的節點分為幾種型別：​

​

| 節點型別 | 說明 | 示例 |​

|----------|------|------|​

| \*\*函式節點\*\* | 執行普通Python邏輯 | 資料處理、條件判斷 |​

| \*\*LLM節點\*\* | 呼叫大模型進行推理 | 文字生成、分析判斷 |​

| \*\*工具節點\*\* | 呼叫外部工具 | 漏洞掃描、API呼叫 |​

| \*\*子圖節點\*\* | 巢狀執行另一個圖 | 複雜流程的模組化 |​

​

\### 2）函式節點 vs LLM節點​

​

\*\*函式節點\*\*：執行確定性邏輯，不涉及LLM呼叫。​

​

LLM節點：呼叫大模型進行推理和生成。​

​

Code block​

Python

\# llm\_node.py​

from langchain\_deepseek import ChatDeepSeek​

from langchain\_core.messages import SystemMessage, HumanMessage​

​

\# 初始化DeepSeek模型​

llm = ChatDeepSeek(​

model="deepseek-chat",​

api\_key="your\_api\_key\_here",​

temperature=0.3 \# 安全場景使用較低溫度，確保輸出穩定​

)​

​

​

3）工具節點的整合​

工具節點允許圖中的節點呼叫外部工具和API。​

​

Code block​

Python

​

2.4 邊（Edge）​

1）普通邊：順序執行​

普通邊定義了節點之間的順序執行關係。​

​

Code block​

Python

\# normal\_edge.py​

\# 新增普通邊：analyzer執行完後執行executor​

workflow.add\_edge("analyzer", "executor")​

​

\# 設定入口點​

workflow.set\_entry\_point("analyzer")​

​

\# 設定結束點​

workflow.add\_edge("reporter", END)​

​

2）條件邊：動態路由​

條件邊根據狀態中的值動態決定下一個執行的節點。​

​

Code block​

Python

\# conditional\_edge.py​

def route\_after\_analysis(state: SecurityAnalysisState) -> str:​

"""分析後的路由邏輯"""​

\# 如果有高危漏洞，需要人工稽核​

high\_vulns = \[v for v in state.get("vulnerabilities", \[\]) if v.get("severity") == "high"\]​

​

if high\_vulns:​

return "manual\_review"​

​

\# 如果漏洞較多，需要詳細報告​

if len(state.get("vulnerabilities", \[\])) > 10:​

return "detailed\_report"​

​

3）入口點與結束點​

入口點定義了圖的起始節點，結束點（END）標記了圖的終止。​

​

Code block​

Python

\# entry\_exit.py​

from langgraph.graph import StateGraph, END​

​

workflow = StateGraph(SecurityAnalysisState)​

​

\# 設定入口點​

workflow.set\_entry\_point("start\_node")​

​

\# 新增結束邊​

workflow.add\_edge("final\_node", END)​

​

\# 也可以使用條件邊路由到END​

​

三、LangGraph開發實戰​

3.1 環境搭建與安裝​

1）依賴安裝​

​

Code block​

Bash

\# 安裝LangGraph核心包​

pip install langgraph​

​

\# 安裝LangChain整合包​

pip install langchain langchain-core langchain-community​

​

\# 安裝DeepSeek整合包​

pip install langchain-deepseek​

​

\# 安裝其他依賴​

pip install python-dotenv typing-extensions​

​

2）DeepSeek API配置​

​

Code block​

Python

\# config.py​

import os​

from dotenv import load\_dotenv​

​

\# 載入環境變數​

load\_dotenv()​

​

\# DeepSeek API配置​

DEEPSEEK\_API\_KEY = os.getenv("DEEPSEEK\_API\_KEY", "your\_api\_key\_here")​

DEEPSEEK\_BASE\_URL = "https://api.deepseek.com"​

​

\# 模型配置​

MODEL\_CONFIG = {​

"model": "deepseek-chat",​

"temperature": 0.3,​

"max\_tokens": 4096,​

"api\_key": DEEPSEEK\_API\_KEY,​

}​

​

.env檔案：​

​

Code block​

Plain Text

DEEPSEEK\_API\_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx​

​

3.2 第一個LangGraph應用​

1）簡單的線性圖示例​

讓我們從一個最簡單的例子開始——一個能夠回答安全問題的線性圖。​

​

Code block​

Python

​

2）狀態定義與傳遞​

在上面的例子中，我們定義了ChatState作為狀態結構。讓我們更深入地理解狀態的傳遞機制：​

​

Code block​

Python

\# state\_flow.py​

from typing import TypedDict, Annotated, List, Optional​

from langgraph.graph import StateGraph, END​

from langgraph.graph import add\_messages​

from langchain\_core.messages import BaseMessage​

​

class AnalysisState(TypedDict):​

\# 訊息歷史，使用add\_messages自動追加​

messages: Annotated\[List\[BaseMessage\], add\_messages\]​

​

\# 分析階段​

stage: str \# "input" -> "analyze" -> "output"​

​

\# 中間結果​

raw\_data: Optional\[str\]​

analysis\_result: Optional\[str\]​

​

def input\_node(state: AnalysisState) -> dict:​

"""輸入節點：準備資料"""​

return {​

"stage": "analyze",​

"raw\_data": "原始安全資料...",​

"messages": \[("system", "資料準備完成，開始分析")\]​

}​

​

def analyze\_node(state: AnalysisState) -> dict:​

"""分析節點：處理資料"""​

\# 讀取上一個節點的資料​

raw\_data = state.get("raw\_data", "")​

​

​

3）圖的編譯與執行​

編譯是將圖定義轉換為可執行物件的過程：​

​

Code block​

Python

\# compile\_options.py​

from langgraph.checkpoint.memory import MemorySaver​

​

\# 基本編譯​

app = workflow.compile()​

​

\# 帶檢查點的編譯​

checkpointer = MemorySaver()​

app\_with\_checkpoint = workflow.compile(​

checkpointer=checkpointer,​

interrupt\_before=\["human\_review"\], \# 在指定節點前中斷​

interrupt\_after=\["sensitive\_action"\] \# 在指定節點後中斷​

)​

​

\# 執行圖​

\# 方式1：同步執行​

result = app.invoke(initial\_state)​

​

\# 方式2：帶配置執行​

config = {"configurable": {"thread\_id": "session-001"}}​

result = app.invoke(initial\_state, config=config)​

​

\# 方式3：流式執行​

for event in app.stream(initial\_state):​

print(event)​

​

3.3 條件分支與迴圈​

1）條件邊的實現​

條件分支是構建智慧Agent的關鍵，讓Agent能夠根據不同情況做出不同決策。​

​

Code block​

Python

\# conditional\_branch.py​

from typing import TypedDict, Literal​

from langgraph.graph import StateGraph, END​

​

class SecurityCheckState(TypedDict):​

target: str​

scan\_result: str​

risk\_level: str \# "low" | "medium" | "high"​

action\_taken: str​

​

def scan\_node(state: SecurityCheckState) -> dict:​

"""掃描節點"""​

target = state\["target"\]​

\# 模擬掃描邏輯​

if "admin" in target or "root" in target:​

return {"scan\_result": "發現敏感路徑", "risk\_level": "high"}​

elif "api" in target:​

return {"scan\_result": "發現API端點", "risk\_level": "medium"}​

else {"scan\_result": "未發現明顯風險", "risk\_level": "low"}​

​

def route\_by\_risk(state: SecurityCheckState) -> Literal\["high\_risk", "medium\_risk", "low\_risk"\]:​

"""根據風險等級路由"""​

​

2）迴圈邏輯的應用​

迴圈是Agent實現ReAct推理模式的關鍵。​

​

Code block​

Python

\# react\_loop.py​

from typing import TypedDict, Annotated, List​

from langgraph.graph import StateGraph, END​

from langchain\_core.messages import BaseMessage, HumanMessage, SystemMessage​

from langgraph.graph import add\_messages​

from langchain\_deepseek import ChatDeepSeek​

import os​

from dotenv import load\_dotenv​

​

load\_dotenv()​

​

class ReActState(TypedDict):​

messages: Annotated\[List\[BaseMessage\], add\_messages\]​

thoughts: List\[str\]​

actions: List\[str\]​

​

3）ReAct模式的圖實現​

上面的程式碼展示瞭如何用LangGraph實現\*\*ReAct（Reasoning + Acting）\*\*模式：​

ReAct模式的核心迴圈：​

1.

思考（Think）：分析當前狀態，決定下一步行動​

2.

行動（Act）：執行具體的工具呼叫或分析任務​

3.

觀察（Observe）：獲取行動的結果​

4.

迴圈：回到思考，直到獲得足夠資訊​

ReAct vs 傳統Chain對比：​

​



| 特性 | 傳統Chain | ReAct Loop |
| --- | --- | --- |
| 執行方式 | 線性，一次透過 | 迴圈，多次迭代 |
| 決策能力 | 預定義路徑 | 動態決策 |
| 工具使用 | 固定呼叫 | 按需呼叫 |
| 適用場景 | 簡單任務 | 複雜推理任務 |



​

3.4 工具整合​

1）工具節點的定義​

LangGraph提供了ToolNode來簡化工具整合：​

​

Code block​

Python

\# tool\_integration.py​

from langchain\_core.tools import tool​

from langgraph.prebuilt import ToolNode​

from typing import List, Dict​

​

@tool​

def port\_scanner(target: str, ports: str = "1-1000") -> Dict:​

"""掃描目標埠​

​

Args:​

target: 目標IP或域名​

ports: 埠範圍，預設1-1000​

​

Returns:​

掃描結果字典​

"""​

\# 實際應用中呼叫nmap等工具​

return {​

"target": target,​

"open\_ports": \[22, 80, 443\],​

"status": "completed"​

}​

​

@tool​

def vulnerability\_scanner(target: str) -> List\[Dict\]:​

"""掃描目標漏洞​

​

Args:​

target: 目標URL​

​

​

2）工具呼叫與結果處理​

在圖中整合工具節點：​

​

Code block​

Python

​

3）安全工具整合示例​

完整的安全工具整合：​

​

Code block​

Python

​

四、多智慧體協作（Multi-Agent）​

4.1 多智慧體架構設計​

1）角色分工與協作模式​

多智慧體系統的核心在於合理的角色分工和高效的協作機制。​

常見的多智慧體架構模式：​

​



| 模式 | 說明 | 適用場景 |
| --- | --- | --- |
| 主從模式 | 一個Supervisor分配任務給Worker | 任務明確、可分解 |
| 對等模式 | Agent之間平等協作 | 需要多角度分析 |
| 層級模式 | 多級Supervisor層層下發 | 複雜、多層次任務 |
| 流水線模式 | Agent按順序處理 | 需要多階段處理 |



​

2）主從模式 vs 對等模式​

主從模式（Supervisor-Worker）：​

​

Code block​

Plain Text

​

對等模式（Peer-to-Peer）：​

​

Code block​

Plain Text

┌─────────┐ ┌─────────┐​

│ Agent A │◄───►│ Agent B │​

│(漏洞專家)│ │(滲透專家)│​

└────┬────┘ └────┬────┘​

│ │​

└───────┬───────┘​

▼​

┌───────────┐​

│ 共享記憶 │​

└───────────┘​

​

3）通訊機制​

Agent之間的通訊方式：​

​



| 方式 | 說明 | 優點 | 缺點 |
| --- | --- | --- | --- |
| 共享狀態 | 透過State共享資料 | 簡單直觀 | 可能產生衝突 |
| 訊息傳遞 | Agent之間傳送訊息 | 解耦性好 | 需要設計訊息格式 |
| 黑板模式 | 共享的"黑板"儲存資訊 | 靈活性高 | 一致性難保證 |



​

4.2 Supervisor模式​

1）監督者的定義與職責​

Supervisor是多智慧體系統中的排程中心，負責：​

•

接收使用者請求​

•

分析任務需求​

•

選擇合適的Worker​

•

彙總執行結果​

​

Code block​

Python

​

2）任務分配與結果彙總​

更完善的Supervisor實現，支援多輪任務分配和結果彙總：​

​

Code block​

Python

​

3）安全場景中的多智慧體應用​

漏洞分析智慧體團隊：​

​

Code block​

Python

\# vuln\_analysis\_team.py​

from typing import TypedDict, Annotated, List, Dict​

from langgraph.graph import StateGraph, END​

​

4.3 安全場景中的多智慧體應用​

1）威脅情報協作分析​

多源情報融合分析：​

​

Code block​

Python

​

2）自動化應急響應​

安全事件自動化響應流程：​

​

Code block​

Python

\# incident\_response.py​

from typing import TypedDict, Annotated, List, Dict​

from langgraph.graph import StateGraph, END​

from langchain\_core.messages import BaseMessage​

from langgraph.graph import add\_messages​

​

class IncidentResponseState(TypedDict):​

messages: Annotated\[List\[BaseMessage\], add\_messages\]​

incident: Dict \# 安全事件資訊​

triage\_result: Dict \# 分診結果​

containment\_actions: List\[str\] \# 遏制措施​

eradication\_actions: List\[str\] \# 根除措施​

recovery\_actions: List\[str\] \# 恢復措施​

lessons\_learned: str \# 經驗教訓​

​

def triage\_agent(state: IncidentResponseState) -> dict:​

"""分診Agent：評估事件嚴重性"""​

incident = state\["incident"\]​

​

五、高階特性與最佳實踐​

5.1 流式輸出（Streaming）​

1）流式處理的優勢​

流式輸出允許在圖執行過程中實時獲取中間結果，而不是等待整個圖執行完成。​

流式輸出的價值：​

​



| 場景 | 傳統方式 | 流式方式 |
| --- | --- | --- |
| 使用者體驗 | 等待長時間無反饋 | 實時顯示進度 |
| 除錯 | 只能看到最終結果 | 觀察每個步驟 |
| 超時風險 | 長任務容易超時 | 分步處理，降低風險 |



​

2）節點級流式輸出​

​

Code block​

Python

​

3）事件流的訂閱與處理​

​

Code block​

Python

\# event\_stream.py​

from typing import TypedDict, Annotated, List, Any, Dict​

from langgraph.graph import StateGraph, END​

from langchain\_core.messages import BaseMessage​

from langgraph.graph import add\_messages​

​

class EventState(TypedDict):​

messages: Annotated\[List\[BaseMessage\], add\_messages\]​

events: List\[Dict\[str, Any\]\]​

​

def node\_with\_events(state: EventState) -> dict:​

"""帶事件發射的節點"""​

events = \[​

{"type": "progress", "message": "開始處理", "progress": 0},​

{"type": "progress", "message": "處理中", "progress": 50},​

{"type": "progress", "message": "處理完成", "progress": 100}​

\]​

​

return {​

"events": events,​

"messages": \[("system", "節點執行完成")\]​

}​

​

\# 建立圖​

workflow = StateGraph(EventState)​

workflow.add\_node("processor", node\_with\_events)​

workflow.add\_edge("processor", END)​

workflow.set\_entry\_point("processor")​

app = workflow.compile()​

​

\# 訂閱事件流​

for event in app.stream({"messages": \[\], "events": \[\]}):​

​

5.2 持久化與恢復​

1）檢查點的配置​

​

Code block​

Python

​

2）狀態的持久化儲存​

​

Code block​

Python

​

3）應用中斷與恢復​

​

Code block​

Python

​

4）持久化的最佳實踐​

在生產環境中使用持久化時，需要遵循以下最佳實踐：​

​



| 實踐 | 說明 | 適用場景 |
| --- | --- | --- |
| 選擇合適的儲存後端 | 根據資料量和併發需求選擇 | 小規模用SQLite，大規模用PostgreSQL |
| 配置合理的保留策略 | 設定檢查點保留時間和數量 | 避免儲存空間無限增長 |
| 實現狀態序列化 | 確保自定義型別可序列化 | 複雜物件需要自定義序列化 |
| 處理併發訪問 | 使用適當的鎖機制 | 多執行緒/多程序場景 |
| 定期清理歷史檢查點 | 刪除過期的檢查點資料 | 保持儲存整潔 |



​

​

Code block​

Python

​

5.3 人機協作（Human-in-the-Loop）​

1）人工稽核節點的設計​

Human-in-the-Loop是LangGraph的核心特性，讓AI應用能夠在關鍵節點暫停，等待人工確認或輸入。​

​

Code block​

Python

​

2）中斷與繼續機制​

中斷型別：​

​

​

3）安全場景中的應用​

在安全場景中，Human-in-the-Loop非常關鍵：​

​

​

5.4 錯誤處理與重試​

1）異常捕獲機制​

​

Code block​

Python

\# error\_handling.py​

from typing import TypedDict, Annotated, List​

from langgraph.graph import StateGraph, END​

from langchain\_core.messages import BaseMessage​

from langgraph.graph import add\_messages​

​

class RobustState(TypedDict):​

messages: Annotated\[List\[BaseMessage\], add\_messages\]​

retry\_count: int​

max\_retries: int​

error: str​

result: str​

​

def risky\_node(state: RobustState) -> dict:​

"""可能失敗的節點"""​

import random​

​

try:​

\# 模擬可能失敗的操作​

if random.random() < 0.5:​

​

2）自動重試策略​

​

Code block​

Python

​

3）錯誤處理的最佳實踐​

​



| 實踐 | 說明 | 示例 |
| --- | --- | --- |
| 設定最大重試次數 | 避免無限迴圈 | max\_retries: 3 |
| 使用指數退避 | 減少系統壓力 | backoff \* 2^retry\_count |
| 實現降級方案 | 保證基本可用性 | 備用節點或預設值 |
| 記錄錯誤日誌 | 便於問題排查 | 寫入狀態或日誌系統 |
| 區分臨時/永久錯誤 | 決定是否重試 | 網路錯誤可重試，邏輯錯誤不重試 |



​

​

Code block​

Python

\# error\_classification.py​

from enum import Enum​

​

class ErrorType(Enum):​

TEMPORARY = "temporary" \# 臨時錯誤，可重試​

PERMANENT = "permanent" \# 永久錯誤，不可重試​

RATE\_LIMIT = "rate\_limit" \# 限流錯誤，需要等待​

​

def classify\_error(error: Exception) -> ErrorType:​

"""分類錯誤型別"""​

error\_msg = str(error).lower()​

​

if "timeout" in error\_msg or "connection" in error\_msg:​

return ErrorType.TEMPORARY​

elif "rate limit" in error\_msg or "429" in error\_msg:​

return ErrorType.RATE\_LIMIT​

else:​

return ErrorType.PERMANENT​

​

六、安全實戰案例​

6.1 智慧漏洞分析系統​

1）需求分析與架構設計​

系統目標：構建一個能夠自動掃描、分析和報告漏洞的智慧系統。​

架構設計：​

​

Code block​

Plain Text

​

核心元件：​

​



| 元件 | 職責 | 技術實現 |
| --- | --- | --- |
| Supervisor Agent | 任務規劃與排程 | LangGraph條件路由 |
| Scanner Agent | 漏洞掃描 | Nmap/Nuclei整合 |
| Analyzer Agent | 漏洞分析 | DeepSeek大模型 |
| Reporter Agent | 報告生成 | 模板+LLM |



​

2）完整實現程式碼​

​

Code block​

Python

\# vuln\_analysis\_system.py​

​

3）系統最佳化建議​

​



| 最佳化方向 | 具體措施 | 預期效果 |
| --- | --- | --- |
| 並行掃描 | 使用並行邊同時執行多個掃描任務 | 提升掃描效率 |
| 增量更新 | 支援增量掃描，只掃描變化部分 | 減少重複工作 |
| 知識庫整合 | 整合CVE/NVD漏洞庫 | 提升分析準確性 |
| 快取機制 | 快取掃描結果，避免重複掃描 | 提升響應速度 |



​

6.2 自動化安全報告生成​

1）報告生成流程​

​

Code block​

Plain Text

收集資料 → 分析資料 → 生成報告 → 人工稽核 → 釋出報告​

│ │ │ │ │​

▼ ▼ ▼ ▼ ▼​

日誌採集 統計分析 LLM生成 稽核節點 輸出PDF​

​

2）多智慧體協作實現​

​

Code block​

Python

\# data\_collector\_agent.py​

from typing import TypedDict, Annotated, List, Dict, Any​

from langgraph.graph import StateGraph, END​

from langchain\_core.messages import BaseMessage​

from langgraph.graph import add\_messages​

​

class ReportState(TypedDict):​

messages: Annotated\[List\[BaseMessage\], add\_messages\]​

raw\_data: Dict\[str, Any\]​

analysis\_result: Dict\[str, Any\]​

report\_draft: str​

approved: bool​

final\_report: str​

​

def data\_collector\_node(state: ReportState) -> dict:​

"""資料收集節點"""​

\# 收集各類安全資料​

raw\_data = {​

​

6.3 威脅情報處理流水線​

1）流水線架構​

​

Code block​

Plain Text

情報採集 → 情報清洗 → 情報分析 → 情報儲存 → 情報應用​

│ │ │ │ │​

▼ ▼ ▼ ▼ ▼​

多源採集 格式標準化 關聯分析 知識圖譜 告警匹配​

​

2）並行處理實現​

​

Code block​

Python

\# threat\_intel\_pipeline.py​

from typing import TypedDict, Annotated, List, Dict, Any​

from langgraph.graph import StateGraph, END​

from langchain\_core.messages import BaseMessage​

from langgraph.graph import add\_messages​

import asyncio​

​

class ThreatIntelState(TypedDict):​

messages: Annotated\[List\[BaseMessage\], add\_messages\]​

raw\_intel: List\[Dict\[str, Any\]\]​

processed\_intel: List\[Dict\[str, Any\]\]​

analysis\_result: Dict\[str, Any\]​

storage\_status: str​

​

def collect\_from\_source\_a(state: ThreatIntelState) -> dict:​

"""從源A採集情報"""​

return {​

"raw\_intel": \[{"source": "A", "type": "malware", "indicator": "evil.com"}\],​

"messages": \[("system", "源A採集完成")\]​

}​

​

def collect\_from\_source\_b(state: ThreatIntelState) -> dict:​

​

七、LangGraph vs 其他框架​

7.1 框架對比​

​



| 維度 | LangGraph | AutoGen | CrewAI | MetaGPT |
| --- | --- | --- | --- | --- |
| 開發語言 | Python | Python | Python | Python |
| 核心抽象 | 圖結構 | 對話模式 | 角色+任務 | 角色+SOP |
| 狀態管理 | 內建檢查點 | 對話歷史 | 任務狀態 | 專案狀態 |
| 人機協作 | 原生支援 | 支援 | 有限 | 有限 |
| 視覺化 | 圖視覺化 | 對話流 | 任務流 | 流程圖 |
| 學習曲線 | 中等 | 較低 | 低 | 中等 |
| 適用場景 | 複雜工作流 | 對話式協作 | 任務型協作 | 軟體開發 |



​

7.2 選型建議​

​



| 應用場景 | 推薦框架 | 理由 |
| --- | --- | --- |
| 複雜狀態機 | LangGraph | 圖結構天然支援 |
| 對話式Agent | AutoGen | 對話模式更自然 |
| 任務型協作 | CrewAI | 角色定義直觀 |
| 軟體開發 | MetaGPT | 內建SOP和角色 |
| 需要中斷恢復 | LangGraph | 檢查點機制完善 |
| 快速原型 | CrewAI | 上手簡單 |



​

八、總結與展望​

8.1 核心要點回顧​

LangGraph的核心價值：​

​



| 特性 | 價值 | 應用場景 |
| --- | --- | --- |
| 圖結構 | 靈活的控制流 | 複雜業務邏輯 |
| 狀態管理 | 有狀態的執行 | 多輪對話、長任務 |
| 檢查點 | 斷點續傳 | 故障恢復、除錯 |
| 人機協作 | 人工幹預 | 稽核流程、決策點 |
| 多智慧體 | 團隊協作 | 複雜任務分解 |



​

學習路徑建議：​

​

Code block​

Plain Text

​

8.2 未來發展趨勢​

LangGraph的發展方向：​

​



| 方向 | 預期發展 | 影響 |
| --- | --- | --- |
| 效能最佳化 | 更快的圖執行 | 支援更大規模應用 |
| 生態豐富 | 更多預置元件 | 降低開發門檻 |
| 視覺化增強 | 更強的除錯工具 | 提升開發體驗 |
| 雲原生支援 | 原生雲部署 | 簡化運維 |
| 標準統一 | 與其他框架互操作 | 降低遷移成本 |



​

給開發者的建議：​

1.

從小處著手：先用LangGraph實現簡單的ReAct Agent，熟悉圖結構​

2.

善用檢查點：在開發階段就啟用檢查點，便於除錯​

3.

合理使用人機協作：在關鍵決策點設定人工稽核​

4.

關注社群動態：LangGraph更新頻繁，及時跟進新特性​

5.

結合實際場景：選擇最適合的框架，不要為了用而用​

​

​

作者：無涯​