在上一篇,我們已經掌握了 State、Node 與 Edge 的核心觀念。本篇直接進入實作,動手打造第一個 LangGraph 對話程式。對應的範例程式位於 langgraph-chatbot。
在 LangGraph 中,節點(Node)本質上就是一個普通的 Python 函式。它就像流水線上的一個工作站,遵守固定的規範:
我們定義負責與模型溝通的 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 末端。result["messages"] 末端即為模型剛產生的 AIMessage。有了圖的建構器與單輪執行函式後,接著撰寫 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 自動接手跨請求的狀態保存與恢復。