昨天教練有了記憶,但不管你說什麼,它都只會做同一件事:把整段對話丟給模型,換一句回應。使用者說「幫我排計畫」跟「我今天做完了」,走的是完全一樣的路線,教練根本分不出這兩句話該用不同方式處理。
今天要讓圖第一次出現「分岔路」。教練要先聽懂使用者這句話屬於哪一種意圖,再決定接下來要做什麼,這是 LangGraph 真正發揮威力的地方:不再是一條直線,而是一張會依照條件走不同路的圖。
真實世界的意圖分類可以做得很複雜,但 30 天時間有限,先簡化成三種,夠用就好:
判斷意圖這件事,理論上也可以丟給 Ollama 判斷,但今天故意選最簡單的做法:關鍵字匹配。原因很實際:
以後如果三個分類不夠用,或關鍵字誤判太多,隨時可以把 classify_intent 換成呼叫模型判斷,介面完全不用改,這是先簡單再視情況升級的思路。
昨天的圖是一條直線:START → chat → END。今天要變成這樣:
┌──→ plan(計畫) ──┐
START → router ──┼──→ report(回報)──┼──→ END
└──→ qa(問答) ──┘
router 是一個 Node,跟平常的 Node 一樣,收 State、回傳更新後的 State,只是它更新的是一個新欄位 intent。接在 router 後面的不是普通的 add_edge,而是 add_conditional_edges:LangGraph 會讀 intent 這個值,決定接下來要走哪一條路。
檔案位置: backend/graph.py
狀態: 修改檔案(在 ChatState 裡加一個欄位)
用途: State 多記錄一個「目前這輪對話的意圖」
依賴: 無
class ChatState(TypedDict):
"""messages 是一個清單,每輪對話會被「加進去」而不是覆蓋掉"""
messages: Annotated[list, add_messages]
intent: str
檔案位置: backend/router_node.py
狀態: 新增檔案
用途: 用關鍵字匹配判斷使用者這句話屬於哪一種意圖
依賴: 無
PLAN_KEYWORDS = ["計畫", "安排", "規劃", "重新排", "調整進度"]
REPORT_KEYWORDS = ["完成", "做完", "讀完", "學完", "卡住", "遇到問題", "進度落後", "延期"]
def classify_intent(state: dict) -> dict:
"""Node:讀最後一則使用者訊息,用關鍵字判斷屬於哪一種意圖"""
latest_message = state["messages"][-1].content
if any(keyword in latest_message for keyword in PLAN_KEYWORDS):
intent = "plan"
elif any(keyword in latest_message for keyword in REPORT_KEYWORDS):
intent = "report"
else:
intent = "qa"
return {"intent": intent}
邏輯很直白:先檢查有沒有出現「計畫」類的關鍵字,再檢查「回報」類的,兩個都沒中就歸類成「問答」。PLAN_KEYWORDS 放在前面檢查,是因為像「調整進度」這種句子同時帶有計畫和回報的味道,先判斷計畫類,符合大部分使用情境。
Day 13 才會真正寫出完整的 Planner Agent,今天先讓三個目的地都能各自回應、證明路由有走對,之後再把裡面的邏輯換成真正的功能。
檔案位置: backend/graph.py
狀態: 修改檔案(接續步驟1,繼續往下加)
用途: 定義三個意圖各自對應的 Node
依賴: langchain-ollama, langchain-core
from langchain_core.messages import SystemMessage
from langchain_ollama import ChatOllama
_llm = ChatOllama(model="llama3.1:8b", temperature=0.3)
_PLAN_SYSTEM = "使用者想建立或調整學習計畫。先簡短確認你聽懂了需求,計畫細節之後會有專門的功能處理,這裡不用真的生成完整計畫。"
_REPORT_SYSTEM = "使用者在回報進度或遇到的狀況。給予簡短的鼓勵或回應,讓使用者知道你有聽到。"
_QA_SYSTEM = "使用者在問一般性的問題,直接簡短回答。"
def plan_node(state: ChatState) -> dict:
"""處理計畫意圖"""
prompt = [SystemMessage(content=_PLAN_SYSTEM)] + state["messages"]
response = _llm.invoke(prompt)
return {"messages": [response]}
def report_node(state: ChatState) -> dict:
"""處理回報意圖"""
prompt = [SystemMessage(content=_REPORT_SYSTEM)] + state["messages"]
response = _llm.invoke(prompt)
return {"messages": [response]}
def qa_node(state: ChatState) -> dict:
"""處理問答意圖"""
prompt = [SystemMessage(content=_QA_SYSTEM)] + state["messages"]
response = _llm.invoke(prompt)
return {"messages": [response]}
三個 Node 現在做的事很像,都是「換一個系統提示詞、呼叫模型」,差別只在系統提示詞內容不同。Day 13 開始,plan_node 會被換成真正查資料庫、生成結構化計畫的邏輯,其他兩個之後也會陸續補完。
檔案位置: backend/graph.py
狀態: 修改檔案(接續步驟3,取代原本的圖組建邏輯)
用途: 加上 router 與三條分岔路線,組建最終的 graph
依賴: langgraph, langgraph-checkpoint-sqlite
import sqlite3
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from router_node import classify_intent
_conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
_checkpointer = SqliteSaver(_conn)
graph_builder = StateGraph(ChatState)
graph_builder.add_node("router", classify_intent)
graph_builder.add_node("plan", plan_node)
graph_builder.add_node("report", report_node)
graph_builder.add_node("qa", qa_node)
graph_builder.add_edge(START, "router")
# 分岔:讀 State 裡的 intent,決定接下來走哪個 Node
graph_builder.add_conditional_edges(
"router",
lambda state: state["intent"],
{"plan": "plan", "report": "report", "qa": "qa"},
)
graph_builder.add_edge("plan", END)
graph_builder.add_edge("report", END)
graph_builder.add_edge("qa", END)
graph = graph_builder.compile(checkpointer=_checkpointer)
add_conditional_edges("router", lambda state: state["intent"], {...}) 是今天的重點:第一個參數是分岔點的 Node 名稱,第二個參數是一個函式,讀 State 回傳一個字串,第三個參數是「字串對應到哪個 Node」的對照表。跑到 router 之後,LangGraph 會呼叫這個函式,拿到 intent 的值,查表決定走哪一條路。
完整的 graph.py 到這裡長這樣:
import sqlite3
from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import SystemMessage
from langchain_ollama import ChatOllama
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from router_node import classify_intent
class ChatState(TypedDict):
"""messages 是一個清單,每輪對話會被「加進去」而不是覆蓋掉"""
messages: Annotated[list, add_messages]
intent: str
_llm = ChatOllama(model="llama3.1:8b", temperature=0.3)
_PLAN_SYSTEM = "使用者想建立或調整學習計畫。先簡短確認你聽懂了需求,計畫細節之後會有專門的功能處理,這裡不用真的生成完整計畫。"
_REPORT_SYSTEM = "使用者在回報進度或遇到的狀況。給予簡短的鼓勵或回應,讓使用者知道你有聽到。"
_QA_SYSTEM = "使用者在問一般性的問題,直接簡短回答。"
def plan_node(state: ChatState) -> dict:
"""處理計畫意圖"""
prompt = [SystemMessage(content=_PLAN_SYSTEM)] + state["messages"]
response = _llm.invoke(prompt)
return {"messages": [response]}
def report_node(state: ChatState) -> dict:
"""處理回報意圖"""
prompt = [SystemMessage(content=_REPORT_SYSTEM)] + state["messages"]
response = _llm.invoke(prompt)
return {"messages": [response]}
def qa_node(state: ChatState) -> dict:
"""處理問答意圖"""
prompt = [SystemMessage(content=_QA_SYSTEM)] + state["messages"]
response = _llm.invoke(prompt)
return {"messages": [response]}
_conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
_checkpointer = SqliteSaver(_conn)
graph_builder = StateGraph(ChatState)
graph_builder.add_node("router", classify_intent)
graph_builder.add_node("plan", plan_node)
graph_builder.add_node("report", report_node)
graph_builder.add_node("qa", qa_node)
graph_builder.add_edge(START, "router")
# 分岔:讀 State 裡的 intent,決定接下來走哪個 Node
graph_builder.add_conditional_edges(
"router",
lambda state: state["intent"],
{"plan": "plan", "report": "report", "qa": "qa"},
)
graph_builder.add_edge("plan", END)
graph_builder.add_edge("report", END)
graph_builder.add_edge("qa", END)
graph = graph_builder.compile(checkpointer=_checkpointer)
跟 Day 11 的版本比起來,差別是:State 多了 intent 欄位、多了 router 和三個目的地 Node、還有 add_conditional_edges 這條分岔路線。ChatOllama 只需要建立一次,三個 Node 共用同一個 _llm,不用重複建立連線。
Router 是純關鍵字比對,不用呼叫模型,所以可以獨立測試,跑得又快又不用等。
檔案位置: backend/test_router.py
狀態: 新增檔案
用途: 用至少10個測試用例驗證意圖分類正確
依賴: langchain-core, router_node
from langchain_core.messages import HumanMessage
from router_node import classify_intent
TEST_CASES = [
("幫我重新規劃這週的計畫", "plan"),
("我想調整一下學習計畫", "plan"),
("幫我安排明天要學的內容", "plan"),
("這週進度需要重新排一下", "plan"),
("我今天的任務做完了", "report"),
("我卡住了,這個概念看不懂", "report"),
("進度有點落後,需要延期", "report"),
("我今天完成了EC2的章節", "report"),
("EC2跟S3有什麼差別?", "qa"),
("什麼是VPC?", "qa"),
("這個服務適合什麼場景使用", "qa"),
]
def main() -> None:
passed = 0
for text, expected in TEST_CASES:
state = {"messages": [HumanMessage(content=text)], "intent": ""}
result = classify_intent(state)
actual = result["intent"]
mark = "✓" if actual == expected else "✗"
if actual == expected:
passed += 1
print(f"{mark} 輸入:「{text}」 預期:{expected} 實際:{actual}")
print(f"\n通過 {passed}/{len(TEST_CASES)} 個測試用例")
if __name__ == "__main__":
main()
執行:
python test_router.py
應該看到全部 11 個測試用例都打勾,最後印出 通過 11/11 個測試用例。
檔案位置: backend/test_graph.py
狀態: 修改檔案(取代 Day 10 的內容)
用途: 驗證三種意圖真的會被路由到不同的 Node
依賴: graph
from graph import graph
def ask(user_message: str, thread_id: str) -> None:
config = {"configurable": {"thread_id": thread_id}}
result = graph.invoke(
{"messages": [{"role": "user", "content": user_message}], "intent": ""},
config,
)
print(f"輸入:{user_message}")
print(f"判斷意圖:{result['intent']}")
print(f"回應:{result['messages'][-1].content}\n")
def main() -> None:
ask("幫我重新規劃這週的學習計畫", thread_id="test-plan")
ask("我今天把EC2的章節讀完了", thread_id="test-report")
ask("VPC是什麼?", thread_id="test-qa")
if __name__ == "__main__":
main()
執行:
python test_graph.py
應該看到三次呼叫的 判斷意圖 分別是 plan、report、qa,而且每一個的回應內容也對得上:計畫那句會確認需求,回報那句會給鼓勵,問答那句會直接回答 VPC 是什麼。這裡故意用三個不同的 thread_id,讓三段對話互不干擾,方便看清楚每種意圖各自的結果。
關鍵字匹配本來就是簡化版做法,遇到一句話同時帶兩種意圖的關鍵字時,只會照 classify_intent 裡檢查的順序取第一個符合的。目前是先檢查 PLAN_KEYWORDS,所以這句應該會被判成 plan;如果實際測試發現常常誤判,調整關鍵字清單或檢查順序即可。
add_conditional_edges 的字典 key 對不上,跑起來報錯{"plan": "plan", "report": "report", "qa": "qa"} 這個字典的 key 必須跟 classify_intent 回傳的字串完全一致,value 則要對應到 add_node 時取的名字。兩邊有一個打錯字就會找不到對應的 Node。
故意的。今天的重點是「路由走得對不對」,不是「每個功能做得多完整」。先確認分岔邏輯正確,之後每天疊加一個 Node 的真正功能(Day 13 的計畫生成、Day 17 的每日建議),骨架不用重組。
test_router.py 不用開 Ollama 也能跑,這樣測試有意義嗎?有意義,而且這是刻意設計的。classify_intent 本身不呼叫模型,純粹是字串比對,獨立測試可以跑得很快、不受模型速度影響,也能放進之後 Day 28 要寫的自動化測試裡,不用每次測都等模型跑。
不會,因為 Checkpoint 是照 thread_id分開存的。同一個使用者如果都用同一個 thread_id,三種意圖的對話歷史其實是接在同一條時間軸上的,這是預期行為,因為現實中一個人本來就會在同一段對話裡問問題、也回報進度、也調整計畫。
今天讓教練第一次「聽懂」使用者在說什麼。寫了 router_node.py 用關鍵字判斷三種意圖,圖上第一次出現分岔,11 個測試用例全部通過,也用 test_graph.py 證明三種意圖真的會被送到不同的 Node 處理。
系統現在是這樣的:
Day 1 ✓ 產品定義完成
Day 2 ✓ 開發環境準備
Day 3 ✓ 專案架構設計
Day 4 ✓ 資料庫設計
Day 5 ✓ SQLite 資料庫建置
Day 6 ✓ FastAPI 基礎
Day 7 ✓ 使用者檔案 API
Day 8 ✓ 理解 LLM Agent 的本質
Day 9 ✓ 連接 Ollama 本機模型
Day 10 ✓ LangGraph 最小範例
Day 11 ✓ Thread 與 State 管理
Day 12 ✓ 簡化的意圖路由(今天)
Day 13 ⬜ 計畫生成 Agent(Planner)
今天的 plan_node 只是一個會回話的空殼。明天要把它換成真正的 Planner Agent,讓它讀懂使用者的目標和可用時間,生成一份有結構的多週學習計畫。