在先前的文章中，筆者探討了如何建立一個 AI 客服助理的基礎架構。然而，一個真正聰明的助理不應該只是被動地執行工具，它需要具備判斷力：判斷何時該直接回答、何時該使用工具、以及工具回傳的資訊是否足以回答使用者的問題。

本篇文章將深入介紹如何使用 LangGraph 來設計一個更進階的對話流程。此範例將示範如何建立一個包含「前置判斷」與「後置審核」機制的 Agent，讓它在與使用者互動時，能做出更精準的決策。

上下文管理，歷史資料取得

在客服機器人中，對話的上下文處理是關鍵。最初的策略是：
直接透過 line_id 取得該使用者 24 小時內的所有對話紀錄，並且依照輪數來切分上下文（例如對話開頭一定是使用者 input）。

👉 問題是：如果使用者在 25 小時前開始問問題，那段對話就會被切掉，導致上下文遺失。

後來筆者改成了 會話過期機制：

• 如果同一個對話 session（cid）內使用者有新的訊息，會延長該會話的過期時間。
• 如果長時間無互動，對話就會過期，不再繼續保留。
• 如果是自建聊天平台，也可以嘗試「對話視窗關閉後就不再紀錄」，讓資料量更輕量。

另外，之前有嘗試使用 Redis + 傳統長期資料庫，Redis 負責快取，資料庫做長期保存。不過後來覺得多維護一個 Redis 蠻麻煩的，目前是使用 Firestore + 過期機制來實踐。

FirebaseManager

筆者寫了一個 FirebaseManager，統一處理 Firebase 的讀寫：

import firebase_admin
from firebase_admin import credentials, firestore, storage
import uuid
from google.cloud.firestore_v1.base_query import FieldFilter
import pytz
from datetime import datetime, timedelta

class FirebaseManager:

    def __init__(self):
        self.cred = credentials.Certificate("serviceAccountKey.json")
        self.app_options = {"projectId": "自己的專案 id", "storageBucket": "自己的 storageBucket"}
        if not firebase_admin._apps:
            firebase_admin.initialize_app(self.cred, self.app_options)
        self.chat_ref = firestore.client().collection("chat")
        self.cid_ref = firestore.client().collection("cid")

    # 取得該使用者當前的 cid ，或是建立新的
    def get_cid_or_create(self, line_id: str) -> str:
        ...

    # 透過 cid 新增對話紀錄
    def add_chat(self, cid: str, line_id: str, role: str, content: str, meta: dict | None = None):
        ...

    # 透過 cid 取得對話紀錄
    def get_chat(self, cid: str) -> list:
        ...

firebase_manager = FirebaseManager()

如果是使用者傳圖片，會先把圖片存到 Firebase Storage，然後把公開連結存到 chat：

firebase_manager.add_chat(
    cid=firebase_manager.get_cid_or_create(line_id),
    line_id=line_id,
    role="user",
    content=f"image：{image_url}"
)

LangGraph 流程設計

LangGraph 可以把對話流程用「狀態圖」來表達，每個節點（node）對應一個動作，節點之間的連線（edge）則代表執行順序或條件分支。

Graph 的重要概念：

• node：任何要做的事情，例如使用工具
• edge：連接節點的線，定義節點的執行方向，也可以條件判斷

1. 定義狀態（State）

State 是共享的上下文容器，每個節點都能讀取或更新其中的值：

from typing import TypedDict
from langgraph.graph.message import add_messages

class GraphState(TypedDict):
    user_id: str
    user_name: str
    cid: str
    question: str
    messages: Annotated[list[BaseMessage], add_messages]
    payload: Optional[Dict[str, Any]]  # 工具輸出的解析
    answer: Optional[str]
    # 其他需要的資訊 ...

2. 撰寫節點

以下為簡略版，可依需求調整：

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from app.service.firebase_service import FirebaseManager
from datetime import datetime
from zoneinfo import ZoneInfo
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage, BaseMessage
from pydantic import BaseModel
from typing import Optional, Literal, List, Dict, Any
import json

# QC 最後一關：協助判斷是否可以回答使用者，或需要追問
class JudgeOut(BaseModel):
    decision: Literal["answer", "followup"]
    final_text: Optional[str] = None
    followup_question: Optional[str] = None

def build_graph(firebase_manager: FirebaseManager, model_name: str = "gemini-2.5-flash"):
    llm = init_chat_model(model_name, model_provider="google_genai", temperature=0, top_p=1.0)

    # --- 1) 載歷史 ---
    def load_history(state: GraphState):
        # 透過 cid 取得歷史紀錄後，組成 LangChain 所需格式
        history_msgs: List[BaseMessage] = []
        return {"messages": history_msgs}
    
    # --- 2) 計畫並發出工具呼叫 ---
    all_tools = [faq_tool, activity_tool]

    def plan_and_call(state: GraphState):
        taiwan_tz = ZoneInfo("Asia/Taipei")
        now_in_taiwan = datetime.now(taiwan_tz)
        formatted_time_chinese = now_in_taiwan.strftime("%Y年%m月%d日 %H點%M分%S秒")

        rules = (
            f"當前時間為 {formatted_time_chinese}。\n"
            "路由規則核心原則：\n"
            "1.禁止在沒有足夠資訊的情況下猜測或編造參數。\n"
            "# --- 工具路由規則 ---\n"
            "- FAQ (如會員、折價券) → get_faq\n"
            "- 活動規則與內容詢問 → get_activity\n"
            "# --- 其他重要規則 ---\n"
            "- 當使用者意圖模糊、缺少必要參數 (如取消訂單未提供單號) 或無法匹配任何工具 → 直接回覆澄清問題。\n"
        )

        msgs: List[BaseMessage] = [SystemMessage(content=rules), *state.get("messages", [])]
        ai = llm.bind_tools(all_tools).invoke(msgs)
        return {"messages": [ai]}

    # --- 3) 預檢門 ---
    def pre_exec_gate(state: GraphState):
        last_ai = next((m for m in reversed(state["messages"]) if isinstance(m, AIMessage)), None)
        if not last_ai:
            return {}
        if not getattr(last_ai, "tool_calls", None):
            text = (last_ai.content or "").strip()
            if text:
                return {"answer": text}
        return {}

    def route_after_pre_exec_gate(state: GraphState):
        return "save_messages" if state.get("answer") else "run_tools"

    # --- 4) 解析/正規化工具輸出 ---
    def collect_payload(state: GraphState):
        tool_msg = next((m for m in reversed(state.get("messages", [])) if isinstance(m, ToolMessage)), None)
        payload = {}  # 代碼省略，實作工具回傳解析
        return {"payload": payload}

    # --- 5) 審核與回答 ---
    def judge_and_compose(state: GraphState):
        pl = state.get("payload") or {}
        tool_name = pl.get("tool_name") or "unknown"
        data = pl.get("data")
        judge_prompt = (
            "你是客服審核與出話助手。根據工具資料判斷是否足以正確回答使用者，"
            "不可編造；不足則精準追問；片段較多則整合成精簡答案。\n"
            "- 若可直接回：decision='answer'，final_text ≤120字。\n"
            "- 若不足：decision='followup'，followup_question 一句話（只問最關鍵）\n"
        )

        msgs = [
            SystemMessage(content=judge_prompt),
            HumanMessage(content=f"使用者問題：{state['messages'][-3:]}"),
            HumanMessage(content=f"工具: {tool_name}"),
            HumanMessage(content=f"工具資料 data（可為清單或字典）：{json.dumps(data, ensure_ascii=False)[:3000]}"),
        ]

        judge = llm.with_structured_output(JudgeOut).invoke(msgs)
        if judge.decision == "followup" and judge.followup_question:
            return {"answer": judge.followup_question}
        return {"answer": judge.final_text}

    # --- 6) 儲存對話紀錄 ---
    def save_messages(state: GraphState):
        # 將對話紀錄存到 Firestore
        return {}

3. Graph 組圖

graph = StateGraph(GraphState)

graph.add_node("load_history", load_history)
graph.add_node("plan_and_call", plan_and_call)
graph.add_node("pre_exec_gate", pre_exec_gate)
graph.add_node("run_tools", run_tools)
graph.add_node("collect_payload", collect_payload)
graph.add_node("judge_and_compose", judge_and_compose)
graph.add_node("save_messages", save_messages)

graph.add_edge(START, "load_history")
graph.add_edge("load_history", "plan_and_call")
graph.add_edge("plan_and_call", "pre_exec_gate")
graph.add_conditional_edges(
    "pre_exec_gate",
    route_after_pre_exec_gate,
    {"run_tools": "run_tools", "save_messages": "save_messages"}
)
graph.add_edge("run_tools", "collect_payload")
graph.add_edge("collect_payload", "judge_and_compose")
graph.add_edge("judge_and_compose", "save_messages")
graph.add_edge("save_messages", END)

compiled_graph = graph.compile()

3. 工具調用範例

這邊先貼實際執行的 code，可依照需求決定向量資料庫與實踐

class FaqToolInput(BaseModel):
    """用自然語言查 FAQ，模型會傳入使用者問題。"""
    question: str = Field(..., description="User's question in natural language")
    top_k: int = Field(4, ge=1, le=10, description="Number of results to retrieve (default 4)")

@tool(name_or_callable="get_faq", args_schema=FaqToolInput, return_direct=True)
def faq_tool(question: str, top_k: int = 4) -> Dict[str, Any]:
    """
    查詢常見問題（FAQ）。
    輸入使用者問題，回傳最相關 FAQ 條目（含問答與摘要）。
    無相關資料直接回答「沒有相關資料」。
    """
    results = []  # 向量資料庫查詢結果
    return {"results": results}

4. 使用範例

from app.graph.assist_graph import build_graph
from app.service.firebase_service import firebase_manager

graph = build_graph(firebase_manager)

result = graph.invoke({
    "user_id": user_id,
    "user_name": user_name,
    "line": user_id,
    "cid": cid,
    "question": question
})

print(result)

結論

透過這樣的設計，客服助理不再只是單純的工具執行者，而是具備 思考 → 行動 → 反思 的能力。筆者的經驗是，這樣的結構能大幅提升回覆品質，減少錯誤回答的情況，並且讓整個對話流程更貼近真人客服的思考方式。