iT邦幫忙

2026 iThome 鐵人賽

DAY 5
0
AI Engineering

AI Agent 系統開發 30 天系列 第 5

第一個 LangGraph 對話程式

  • 分享至 

  • xImage
  •  

在上一篇,我們已經掌握了 State、Node 與 Edge 的核心觀念。本篇直接進入實作,動手打造第一個 LangGraph 對話程式。對應的範例程式位於 langgraph-chatbot。

步驟一:撰寫模型節點函式

在 LangGraph 中,節點(Node)本質上就是一個普通的 Python 函式。它就像流水線上的一個工作站,遵守固定的規範:

  1. 輸入:接收當前圖中的狀態(State)。
  2. 處理:專注做一件事(例如呼叫模型)。
  3. 輸出:回傳狀態更新,表示要更新 State 的哪些欄位。

我們定義負責與模型溝通的 chatbot 節點,也就是一個 Python 函式:

from langgraph.graph import MessagesState

def chatbot(state: MessagesState):
    # 1. 從 State 讀取目前所有的對話訊息歷史
    messages = state["messages"]

    # 2. 呼叫聊天模型產生回覆
    response = model.invoke(messages)

    # 3. 回傳要寫回 State 的新訊息
    return {"messages": [response]}

函式接收的 state 物件包含目前累積的對話歷史 state["messages"]。當節點回傳 {"messages": [response]} 時,LangGraph 會自動把這筆新訊息追加到對話紀錄末端(底層透過 Reducer 機制處理),不會覆蓋既有紀錄,因此不需要手動做 append。

步驟二:組裝與編譯狀態圖

寫好了工作節點後,我們需要一張「圖(Graph)」把節點串起來。圖是整個流程狀態的藍圖,負責定義起點、終點以及節點之間的執行路徑。

我們撰寫 build_graph 函式,將剛才的節點組裝成可執行的圖:

from langgraph.graph import END, START, MessagesState, StateGraph


def build_graph(model):
    # 將節點定義在內部,以便閉包存取傳入的 model
    def chatbot(state: MessagesState):
        response = model.invoke(state["messages"])
        return {"messages": [response]}

    # 1. 建立 StateGraph 藍圖,指定在圖中流動的狀態型別
    builder = StateGraph(MessagesState)

    # 2. 註冊節點(把剛剛寫好的函式掛載進圖中)
    builder.add_node("chatbot", chatbot)

    # 3. 連接邊(設定執行路徑:START -> chatbot -> END)
    builder.add_edge(START, "chatbot")
    builder.add_edge("chatbot", END)

    # 4. 編譯成可被呼叫執行的 CompiledGraph 物件
    return builder.compile()
  • StateGraph(MessagesState)畫設計圖,宣告這張圖要傳遞什麼資料(State),並提供 API 讓我們加入節點與連線。
  • builder.add_node("chatbot", chatbot):將名稱 "chatbot" 與節點函式綁定。
  • builder.compile()將設計圖轉為實體,檢查連線是否合法,並產出一個可以呼叫 invoke() 的執行實體。
  • 為什麼將 model 作為參數傳入?:透過依賴注入(Dependency Injection),讓圖結構與具體模型解耦。正式環境傳入真實的 ChatAnthropic,測試時則可以換成測試模型。

步驟三:撰寫單輪執行函式

完成後,接著撰寫 run_turn 函式,透過 graph.invoke() 執行一輪對話:

from langchain_core.messages import AnyMessage, HumanMessage


def run_turn(
    graph,
    messages: list[AnyMessage],
    question: str,
) -> list[AnyMessage]:
    # 1. 傳入既有歷史與新提問,執行整張圖
    result = graph.invoke(
        {
            "messages": [
                *messages,
                HumanMessage(content=question),
            ]
        }
    )
    # 2. 回傳加入模型回覆後的完整訊息串列
    return result["messages"]

graph.invoke() 接收輸入資料並執行整張狀態圖:

  • 輸入:傳入符合 MessagesState 結構的初始狀態,包含既有歷史與本次新提問。
  • 執行chatbot 節點呼叫模型產生回覆,LangGraph 自動將這筆新回覆追加到 messages 末端。
  • 輸出:執行結束後回傳最終 State,result["messages"] 末端即為模型剛產生的 AIMessage

步驟四:組裝 CLI 多輪對話迴圈

有了圖的建構器與單輪執行函式後,接著撰寫 CLI 入口點,負責初始化模型、編譯圖,並在終端機中透過迴圈維持多輪對話:

import os
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import AIMessage, AnyMessage

def main() -> None:
    if not os.getenv("ANTHROPIC_API_KEY"):
        raise SystemExit("請先設定 ANTHROPIC_API_KEY")

    # 1. 初始化模型
    model = ChatAnthropic(
        model=os.getenv("ANTHROPIC_MODEL", "claude-haiku-4-5"),
        max_tokens=1024,
    )
    # 2. 編譯圖
    graph = build_graph(model)

    # 3. 呼叫端維護的對話歷史串列
    messages: list[AnyMessage] = []

    print("輸入 /exit 結束對話。")

    # 4. 多輪對話迴圈
    while True:
        try:
            question = input("你:").strip()
        except EOFError:
            print()
            break

        if question == "/exit":
            break
        if not question:
            continue

        # 執行一輪對話並更新歷史變數
        messages = run_turn(graph, messages, question)
        response = messages[-1]
        if isinstance(response, AIMessage):
            print(f"AI:{response.text}")


if __name__ == "__main__":
    main()

迴圈的核心是 messages = run_turn(graph, messages, question):每次問答後將包含 AI 回覆的最新串列存回 messages 變數,下一輪再作為歷史傳入,以此在記憶體中維持連續對話。


步驟五:實際執行與終端機互動

切換至範例專案目錄並啟動程式:

uv run langgraph-chatbot

啟動後即可在終端機進行連續問答,輸入 /exit 離開:

輸入 /exit 結束對話。
你:Python 要怎麼在 list 裡面新增東西?
AI:可以使用 append() 方法,例如:fruits.append("蘋果")。
你:那要怎麼把它刪除?
AI:可以使用 remove("蘋果") 刪除指定元素,或是用 pop() 依索引刪除。
你:/exit

這個範例完成了最基礎的 LangGraph 對話骨架:路徑固定為 START → chatbot → END,並由呼叫端的 messages 變數在記憶體中手動維持上下文。

但只要終端機結束或伺服器重啟,這些暫存在變數中的對話紀錄就會全部遺失。下一篇 [[讓 LangGraph 接手對話狀態管理]] 將為 Graph 掛載持久化檢查點(Checkpointer),透過對話 ID(thread_id)由 LangGraph 自動接手跨請求的狀態保存與恢復。


上一篇
用 LangGraph 狀態圖管理 Agent 執行流程
下一篇
讓 LangGraph 接手對話狀態管理
系列文
AI Agent 系統開發 30 天8
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言