Day 5: LangGraph 核心概念介紹

昨天我們建立了一個簡單的對話機器人，今天讓我們深入了解 LangGraph - 一個革命性的 AI 工作流程框架！LangGraph 讓我們能夠以圖形化的方式設計複雜的 AI 應用，就像設計流程圖一樣直觀。

🌟 什麼是 LangGraph？

LangGraph 是 LangChain 生態系統中的一個強大工具，它將 AI 應用的執行流程建模為有向圖（Directed Graph）。每個節點（Node）代表一個特定的功能或操作，邊（Edge）則定義了資料流向和執行順序。

為什麼選擇 LangGraph？

• 🎯 視覺化設計：將複雜邏輯轉化為清晰的圖形結構
• 🔄 狀態管理：自動處理資料流和狀態傳遞
• 🚀 並行處理：支援多個節點同時執行
• 🛠 易於除錯：每個節點都可以獨立測試和監控
• 📈 可擴展性：輕鬆添加新功能而不影響現有結構

🏗 核心概念解析

1. 圖 (Graph)

from langgraph.graph import StateGraph

# 創建一個新的圖
workflow = StateGraph(State)

圖是整個工作流程的容器，定義了節點之間的關係和執行順序。

2. 狀態 (State)

from typing import TypedDict, List

class State(TypedDict):
    messages: List[str]
    user_input: str
    current_task: str
    result: str

狀態是在圖中流動的資料結構，包含了執行過程中需要的所有資訊。

3. 節點 (Node)

def process_input(state: State) -> State:
    """處理使用者輸入的節點"""
    user_input = state["user_input"]
    processed_input = user_input.strip().lower()
    
    return {
        **state,
        "current_task": "input_processed",
        "messages": state["messages"] + [f"處理輸入: {processed_input}"]
    }

# 添加節點到圖中
workflow.add_node("input_processor", process_input)

4. 邊 (Edge)

# 添加固定邊
workflow.add_edge("input_processor", "task_classifier")

# 添加條件邊
def decide_next_step(state: State) -> str:
    current_task = state["current_task"]
    if current_task == "question":
        return "answer_question"
    elif current_task == "task":
        return "execute_task"
    else:
        return "default_response"

workflow.add_conditional_edges(
    "task_classifier",
    decide_next_step,
    {
        "answer_question": "question_handler",
        "execute_task": "task_executor", 
        "default_response": "default_handler"
    }
)

🔧 實戰範例：智能助理工作流程

讓我們建立一個簡單但完整的 LangGraph 應用：

from typing import TypedDict, List, Literal
from langgraph.graph import StateGraph, END
import google.generativeai as genai
from dotenv import load_dotenv
import os

# 載入環境設定
load_dotenv()
genai.configure(api_key=os.getenv('GEMINI_API_KEY'))

# 定義狀態結構
class AssistantState(TypedDict):
    user_input: str
    task_type: str
    context: List[str]
    response: str
    confidence: float

# 初始化 Gemini 模型
model = genai.GenerativeModel('gemini-2.5-flash')

def input_analyzer(state: AssistantState) -> AssistantState:
    """分析使用者輸入，判斷任務類型"""
    user_input = state["user_input"]
    
    # 使用 Gemini 分析輸入類型
    analysis_prompt = f"""
    分析以下使用者輸入，判斷任務類型：
    輸入: {user_input}
    
    請回答以下其中一種類型：
    - question: 一般問題詢問
    - calculation: 數學計算
    - creative: 創作需求
    - task: 執行特定任務
    
    只回答類型名稱，不要額外說明。
    """
    
    try:
        response = model.generate_content(analysis_prompt)
        task_type = response.text.strip().lower()
        confidence = 0.8  # 簡化的信心度
    except:
        task_type = "question"
        confidence = 0.5
    
    return {
        **state,
        "task_type": task_type,
        "confidence": confidence,
        "context": state["context"] + [f"任務類型分析: {task_type}"]
    }

def question_handler(state: AssistantState) -> AssistantState:
    """處理一般問題"""
    user_input = state["user_input"]
    context = "\n".join(state["context"])
    
    prompt = f"""
    基於以下上下文回答問題：
    上下文: {context}
    問題: {user_input}
    
    請提供準確、有幫助的回答。
    """
    
    response = model.generate_content(prompt)
    
    return {
        **state,
        "response": response.text,
        "context": state["context"] + ["已回答一般問題"]
    }

def calculation_handler(state: AssistantState) -> AssistantState:
    """處理數學計算"""
    user_input = state["user_input"]
    
    prompt = f"""
    請幫我計算：{user_input}
    
    提供詳細的計算步驟和最終答案。
    """
    
    response = model.generate_content(prompt)
    
    return {
        **state,
        "response": f"📊 計算結果：\n{response.text}",
        "context": state["context"] + ["已完成數學計算"]
    }

def creative_handler(state: AssistantState) -> AssistantState:
    """處理創作需求"""
    user_input = state["user_input"]
    
    prompt = f"""
    創作要求：{user_input}
    
    請發揮創意，提供優質的創作內容。
    """
    
    response = model.generate_content(prompt)
    
    return {
        **state,
        "response": f"🎨 創作內容：\n{response.text}",
        "context": state["context"] + ["已完成創作任務"]
    }

def task_executor(state: AssistantState) -> AssistantState:
    """執行特定任務"""
    user_input = state["user_input"]
    
    prompt = f"""
    任務需求：{user_input}
    
    請提供執行步驟和相關建議。
    """
    
    response = model.generate_content(prompt)
    
    return {
        **state,
        "response": f"⚡ 任務執行：\n{response.text}",
        "context": state["context"] + ["已執行特定任務"]
    }

def route_task(state: AssistantState) -> Literal["question", "calculation", "creative", "task"]:
    """根據任務類型路由到對應處理器"""
    task_type = state["task_type"]
    
    # 映射任務類型到節點名稱
    routing_map = {
        "question": "question",
        "calculation": "calculation", 
        "creative": "creative",
        "task": "task"
    }
    
    return routing_map.get(task_type, "question")

# 建立工作流程圖
def create_assistant_workflow():
    workflow = StateGraph(AssistantState)
    
    # 添加節點
    workflow.add_node("analyzer", input_analyzer)
    workflow.add_node("question", question_handler)
    workflow.add_node("calculation", calculation_handler)
    workflow.add_node("creative", creative_handler)
    workflow.add_node("task", task_executor)
    
    # 設定入口點
    workflow.set_entry_point("analyzer")
    
    # 添加條件路由
    workflow.add_conditional_edges(
        "analyzer",
        route_task,
        {
            "question": "question",
            "calculation": "calculation",
            "creative": "creative", 
            "task": "task"
        }
    )
    
    # 所有處理器都結束到 END
    workflow.add_edge("question", END)
    workflow.add_edge("calculation", END)
    workflow.add_edge("creative", END)
    workflow.add_edge("task", END)
    
    return workflow.compile()

# 測試工作流程
def test_workflow():
    app = create_assistant_workflow()
    
    # 測試案例
    test_cases = [
        "什麼是機器學習？",
        "幫我計算 25 * 18 + 100",
        "寫一首關於春天的短詩",
        "幫我制定一個學習 Python 的計劃"
    ]
    
    for test_input in test_cases:
        print(f"\n{'='*50}")
        print(f"📝 輸入：{test_input}")
        print('='*50)
        
        # 執行工作流程
        result = app.invoke({
            "user_input": test_input,
            "task_type": "",
            "context": [],
            "response": "",
            "confidence": 0.0
        })
        
        print(f"🎯 任務類型：{result['task_type']}")
        print(f"📊 信心度：{result['confidence']}")
        print(f"💬 回應：\n{result['response']}")

if __name__ == "__main__":
    test_workflow()

🎨 視覺化你的工作流程

# 生成工作流程圖片（需要安裝 graphviz）
def visualize_workflow():
    app = create_assistant_workflow()
    
    try:
        # 生成圖形表示
        graph_image = app.get_graph().draw_mermaid()
        print("工作流程圖：")
        print(graph_image)
    except Exception as e:
        print(f"視覺化失敗：{e}")
        print("提示：安裝 pip install graphviz 以啟用視覺化功能")

🚀 進階特性預覽

1. 並行處理

# 多個節點同時執行
workflow.add_edge("input", ["analyzer1", "analyzer2"])

2. 迴圈和遞迴

# 根據條件決定是否重複執行
def should_continue(state):
    return "continue" if state["need_retry"] else "end"

workflow.add_conditional_edges("processor", should_continue)

3. 錯誤處理

def error_handler(state):
    return {"error": "處理失敗", **state}

workflow.add_node("error_handler", error_handler)

🎯 今日總結

今天我們深入學習了 LangGraph 的核心概念，並建立了一個完整的智能助理工作流程。LangGraph 的圖形化思維讓複雜的 AI 邏輯變得清晰易懂，狀態管理機制確保資料流的正確傳遞。

明天我們將使用 LangGraph 建立第一個實際應用，整合更多實用功能。記得運行今天的範例程式碼，體驗圖形化 AI 工作流程的強大威力！