iT邦幫忙

2026 iThome 鐵人賽

DAY 9
0
AI Engineering

地端 AI 建築學系列 第 9

09 案例一:Workflow (3) AI 請假助理實作 (下)

  • 分享至 

  • xImage
  •  

上一篇我們已經掃過 State 變數表、提問腳本,還有 leaveModule.py 裡負責初始化跟改寫提問的 rewriteLeavePrompt()。這篇接著看 checkForm() 之後的驗證邏輯,最後看怎麼把這些節點組裝成一條真的能跑的 workflow。


請假模組核心邏輯(續)— module/leaveModule.py

checkForm():最忙的節點

程式碼

def checkForm(state: State):

    if True in state["isCreatedAtAsk"] and True not in state["isCreatedAtAsked"]:
        return askForClearCreatedAt(state)

    if state['canSubmit'][-1]:
        return submit(state)

    messages = state["messages"][-1].content.replace('「','').replace('」','')

    if "WrongFirstTimePrompt" in messages.strip():
        return askForClearPromptFirstTime(state)
    state['isFirstTimePromptPassed'].append(True)

    checkPrompt = str(promptTemplate[promptIndex]['checkForm']) \
        .replace('{messages}', messages).replace('{today}', today).replace('{now}', now).replace('{leaveTypes}', leaveTypes)
    answer = llm.invoke(checkPrompt)

    try:
        answer_json = json.loads(answer.content.replace('\n',''))
    except:
        answer_json = json.loads(answer.content.replace('\n','') + '}')

    if answer_json['category'] == '無假別' and len(state['category']) == 0:
        return notAvailableLeaveType(state)
    elif (answer_json['startTime'] == '' and len(state['startTime']) == 0) or (answer_json['endTime'] == '' and len(state['endTime']) == 0):
        return askForTime(state)
    elif answer_json['agent'] == '' and (len(state['agent']) == 0 or len(state['agent'][-1]) == 0):
        return noAgent(state)

    startTime, endTime = correctStartTimeAndEndTime(
        answer_json['startTime'] or state['startTime'][-1],
        answer_json['endTime'] or state['endTime'][-1]
    )
    state['startTime'].append(startTime)
    state['endTime'].append(endTime)

    if answer_json['memo']:
        state['memo'].append(answer_json['memo'])
    elif len(state['memo']) == 0:
        return askForMemo(state)

    category = category_search(answer_json['category'], state['staffData'][-1]['leaveTypes'])
    if not category:
        return notAvailableLeaveType(state)
    state['category'].append(category[0]['name'])

    # 特殊假別追問事件發生日
    if [c for c in ["婚", "喪", "產檢", "陪產"] if c in state['category'][-1]]:
        if True not in state['isCreatedAtAsked']:
            return askForCreatedAt(state)

    if answer_json['agent']:
        agent = agent_search(answer_json['agent'], state['staffData'][-1]['agents'])
        if not agent:
            return notAvailableAgent(state)
        state['agent'].append(agent[0]['name'])

    return preview(state)

說明

這個函式是真正的「路口」,把 LLM 吐出來的 JSON 拿來一路檢查,缺什麼就往回丟哪種追問訊息。拿這段程式碼對照範例對話,會很清楚看到每一步在檢查什麼:

  1. 最上面兩個 if 是優先權最高的兩條快速通道:正在等使用者回答「事件發生日」、或是使用者已經按下確認送出,這兩種情況都不需要重新跑一次完整的解析,直接短路處理。
  2. WrongFirstTimePrompt 對應到範例裡「怪怪的」那一步:rewriteLeavePromptFirstTime 這個 Prompt 判斷這句話跟請假無關,就會照規則輸出 [WrongFirstTimePrompt] 這個特殊字串,checkForm 一看到就轉去 askForClearPromptFirstTime(),回一段附範例句型的引導文字,讓使用者換個講法。
  3. 通過第一關之後才真正呼叫 checkForm 這個 Prompt,把句子解析成 JSON,再依序檢查假別、時間、代理人這三個必要欄位缺不缺 — 這就是前一篇說的「資料檢驗」。

再往下走還有幾個重點:時間不會直接採用,會先丟進 correctStartTimeAndEndTime() 校正成公司規定「以半小時為單位」的格式,這是「跟正式系統對接時,必須用確定性邏輯把關」的典型例子——不會讓 LLM 自己決定要捨去到幾分幾秒,而是寫死規則校正。假別要拿去撞 category_search() 比對可用清單,比對不到就回 notAvailableLeaveType();代理人也是同一套邏輯,用 agent_search() 比對中文名或英文名,比對不到就進 notAvailableAgent()

婚假、喪假、產檢假、陪產假這幾種假別,會多觸發一段子迴圈:第一次遇到會強制轉去 askForCreatedAt() 追問事件發生日,把 isCreatedAtAsk 設成 True;下一輪對話進來,checkForm 最上面那個優先權最高的 if 就會先確認這個追問有沒有被回答完(isCreatedAtAsked),沒回答完就繼續逼問,等於是在主迴圈裡面又包了一個小迴圈。一路檢查通過之後,最後才會走到 preview(state),產生假單預覽。


preview():順便偵測 LLM 是不是卡住了

程式碼

def preview(state: State):
    answer = """
    - 假單預覽 -\n
    姓名:""" + state['staffData'][-1]['name'] + """
    員編:""" + state['staffData'][-1]['id'] + """
    假別:""" + state['category'][-1] + """
    開始時間:""" + state['startTime'][-1] + """
    結束時間:""" + state['endTime'][-1] + """
    代理人:""" + state['agent'][-1] + """
    請假事由:""" + state['memo'][-1] + """
    """

    # 特殊假別補上事件發生日
    if [c for c in ["婚", "喪", "產檢", "陪產"] if c in state['category'][-1]] and len(state['createdAt']) > 0:
        answer += """事件發生日:""" + state['createdAt'][-1]

    # 用暫存內容比對前後是否雷同,雷同代表 LLM 卡住了
    state['previewTmp'].append(answer.replace('\n','').replace(' ',''))
    if len(state['previewTmp']) > 1 and state['previewTmp'][-1] == state['previewTmp'][-2]:
        return askForClearPrompt(state)
    return {"messages": [answer]}

說明

這段程式碼把每次產生的預覽文字都存進 previewTmp 這個 list,然後拿這次跟上一次比對:如果兩次產生的預覽字串一模一樣,代表使用者這輪明明講了新東西,但 LLM 卻沒能正確改寫、結果又吐出一模一樣的結果 — 這種情況與其讓使用者一直卡在原地,不如主動偵測到就轉去 askForClearPrompt(),附上範例句型讓使用者換個講法。這是很實用的工程手法:模型的輸出不可能百分之百可靠,那就用程式邏輯去偵測「異常沒有變化」這個訊號,而不是假設模型一定會正確回應。


submit():真正打進公司 EIP 系統的最後一步

程式碼

def submit(state: State):
    attach_path = str('' if len(state['attachPath'][-1]) == 0 else state['attachPath'][-1].split('/')[-1])

    result = submitLeave(
        inputAccount = state['staffData'][-1]['no'],
        department_no = state['staffData'][-1]['department']['id'],
        agent_no = [str(d['no']) for d in state['staffData'][-1]['agents'] if d['name'] == state['agent'][-1] or d['name_en'] == state['agent'][-1]][0],
        memo = state['memo'][-1],
        category_id = [d['id'] for d in state['staffData'][-1]['leaveTypes'] if d['name'] == state['category'][-1]][0],
        leave_at = str(datetime.datetime.strptime(state['startTime'][-1], "%Y-%m-%d %H:%M").strftime('%Y%m%d%H%M')) + str(datetime.datetime.strptime(state['endTime'][-1], "%Y-%m-%d %H:%M").strftime('%Y%m%d%H%M')),
        attach_path = attach_path
    )

    if result['status']:
        wf_no, wf_day, wf_hour = (str(result['data']).split(',') + ['', '', ''])[:3]
        answer = "姓名:{}\n假別:{}\n時間:{} 到 {}\n事由:{}\n代理人:{}\n請假 {} 天 {} 時".format(
            state['staffData'][-1]['name'], state['category'][-1],
            state['startTime'][-1], state['endTime'][-1],
            state['memo'][-1], state['agent'][-1], wf_day, wf_hour
        )
        return {"messages": ["您的假單送出成功!\nWF單號:{}\n{}".format(wf_no, answer)]}
    else:
        return {"messages": ["抱歉,您的假單送出失敗\nWF訊息:{}".format(result['data']['error_msg'])]}

說明

走到這一步,前面該驗證的都驗證過了,submit() 本身完全沒有 LLM 介入,單純就是把 State 裡整理好的資料組成 EIP API 要的格式,呼叫 submitLeave(),再把回傳結果轉成人看得懂的訊息。對照範例對話「送出失敗,WF 訊息:(WF) 同時段同日期」,就是這裡 result['data']['error_msg'] 原封不動被包進回覆訊息裡;使用者接著只講「時間改成 12/31」,這句話會從頭再跑一次 rewriteLeavePromptcheckForm,因為代理人、假別、事由都還在 State 裡沒被清掉,只有時間被更新,所以馬上就能重新產生預覽、重新送出。


輔助函式

程式碼

def correctStartTimeAndEndTime(startTime, endTime):
    """把 LLM 解析出來的時間,校正成公司規定以 :00 / :30 為單位的請假時間"""
    start = datetime.datetime.strptime(startTime, '%Y-%m-%d %H:%M')
    end = datetime.datetime.strptime(endTime, '%Y-%m-%d %H:%M')
    # ...依 HR 規則校正到最近的 :00 或 :30(完整規則可參考原始檔)
    return startTime, endTime

def agent_search(search_term, data_list):
    """代理人模糊搜尋,中英文名皆可比對"""
    search_term = search_term.lower()
    return [d for d in data_list if search_term in d['name'].lower() or search_term in d['name_en'].lower()]

def category_search(search_term, data_list):
    """假別模糊搜尋"""
    search_term = search_term.lower()
    return [d for d in data_list if search_term in d['name'].lower()]

def is_valid_date(date_string: str) -> bool:
    """檢查字串是否為有效的 yyyy/MM/dd 或 yyyy-MM-dd 格式日期"""
    return bool(re.match(r'^\d{4}[-/](0[1-9]|1[0-2])[-/](0[1-9]|[12]\d|3[01])$', date_string))

說明

這四個是 checkForm()rewriteLeavePrompt() 背後真正在做資料校正跟比對的小工具,邏輯在上面兩節已經解釋過:correctStartTimeAndEndTime 負責把時間校正到公司規定的整點或半點;agent_searchcategory_search 負責拿 LLM 判斷出來的字串去跟員工真正可用的清單模糊比對;is_valid_date 則是在事件發生日追問迴圈裡,用來確認使用者回覆的格式對不對。抓到規則後可以照自己公司規定調整細節。


執行入口與 Graph 組裝 — run-gradio-edgeLLM.py

程式碼

from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langchain_ollama import ChatOllama

from model.stateModel import State
from module.leaveModule import rewriteLeavePrompt, checkForm, setLLM as leaveModuleSetLLM

llm = ChatOllama(model="llama3:8b-instruct-q8_0", temperature=0)

def chat(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

def checkChatType(state: State):
    # 全部走請假流程
    return {"chatType": 'leave'}

def route(state: State):
    return "leave"

workflow = StateGraph(state_schema=State)
workflow.add_node(chat)
workflow.add_node(checkChatType)
workflow.add_node(rewriteLeavePrompt)
workflow.add_node(checkForm)

workflow.add_edge(START, "chat")
workflow.add_edge("chat", "checkChatType")
workflow.add_edge("rewriteLeavePrompt", "checkForm")
workflow.add_conditional_edges(
    source="checkChatType", path=route,
    path_map={"leave": "rewriteLeavePrompt", "end": END}
)

graph = workflow.compile(checkpointer=MemorySaver())

async def Chat(message, history, request):
    thread_id = request.session_hash
    config = {"configurable": {"thread_id": thread_id}}

    leaveModuleSetLLM(llm, 'llama3', str(datetime.date.today()), str(datetime.datetime.now().strftime("%H:%M")), leaveTypes)

    result = await graph.ainvoke(
        {
            "messages": [HumanMessage(message)],
            "chatType": 'leave',
            "category": [], "startTime": [], "endTime": [],
            "agent": [], "memo": [],
            "staffData": [staffData],
            "canSubmit": [False], "attachPath": [''],
            "createdAt": [], "isCreatedAtAsk": [False], "isCreatedAtAsked": [False],
            "previewTmp": [], "isFirstTimePromptPassed": [False]
        },
        config=config
    )
    yield result["messages"][-1].content

說明

這支檔案做兩件事:組 LangGraph 的圖、把圖包成 Gradio 的聊天介面(Gradio 介面的樣式、CSS、範例句這些裝飾跟 workflow 邏輯無關,這裡省略)。temperature=0 這個設定很關鍵,這支機器人不是要 LLM 天馬行空發揮創意,而是要它老老實實照著 Prompt 的規則把話改寫成固定格式,所以溫度直接壓到 0,盡量減少亂跑的機會。

Chat() 是 Gradio 每收到一句話就會呼叫的進入點,這裡藏著整支機器人「聽得懂上下文」的關鍵:thread_id + MemorySaver。每次使用者講一句話,Chat() 都會重新呼叫一次 graph.ainvoke(),理論上是全新的一次呼叫,但因為帶著同一個 thread_id,LangGraph 的 checkpointer 會把上一輪結束時的 State 整包接回來,再疊加這次新輸入的內容繼續跑。所以前一篇說的「局部修改,不用整句重講」,並不是靠 LLM 有記憶力,而是靠這個機制:State 本身沒有不見,使用者這次只講了「代理人改成 XXX」,rewriteLeavePrompt 節點在組 Prompt 的時候,就會把假別、時間這些「這次沒提到」的欄位,用 State 裡存的上一輪的值當預設值補上去。

值得特別提一次的是「反覆追問迴圈」到底藏在哪:看 workflow.add_edge(...) 這幾行會發現,目前程式碼裡的邊其實是一直線的(START → chat → checkChatType → rewriteLeavePrompt → checkForm),並沒有任何流程從 checkForm 繞回 rewriteLeavePrompt,所以這支程式實作的反覆追問迴圈,其實是跨越多次 graph.ainvoke() 呼叫完成的:每一輪對話都是一次全新、單向跑完的 Graph 執行,但因為 thread_id + MemorySaver 把 State 保留住了,所以下一輪呼叫等於是「接著上次結束的地方繼續」。

最後一個小細節:message 裡如果有 [LeaveFormSubmit] 標記字串,前端會在使用者按下「確認送出」按鈕時,把這個標記字串當成訊息傳進來,rewriteLeavePrompt 節點會直接偵測到這個標記,跳過重新解析的流程直接進入送單。


小結

把四支檔案串起來看,這個請假機器人的實作精神大致是這樣:

  • State 用 list 疊加而不是覆蓋值,讓「這次沒提到的欄位」可以自然沿用上一輪的結果,這是局部修改體驗的地基。

  • Prompt 不奢求模型全知全能,地端小模型能力有限,就用大量同義詞規則、格式範例去補,把一部分理解工作用人工規則做掉。

  • LLM 的輸出永遠不能完全信任:預覽文字前後比對、假別/代理人白名單比對,這些工程解邏輯都是在幫 LLM 的不確定性優化。


上一篇
08 案例一 : Workflow (2) AI 請假助理實作 (上)
下一篇
10 案例二:視覺應用(1)看得懂產品的地端 AI 助手
系列文
地端 AI 建築學13
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言