上一篇我們已經掃過 State 變數表、提問腳本,還有
leaveModule.py裡負責初始化跟改寫提問的rewriteLeavePrompt()。這篇接著看checkForm()之後的驗證邏輯,最後看怎麼把這些節點組裝成一條真的能跑的 workflow。
module/leaveModule.pycheckForm():最忙的節點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 拿來一路檢查,缺什麼就往回丟哪種追問訊息。拿這段程式碼對照範例對話,會很清楚看到每一步在檢查什麼:
WrongFirstTimePrompt 對應到範例裡「怪怪的」那一步:rewriteLeavePromptFirstTime 這個 Prompt 判斷這句話跟請假無關,就會照規則輸出 [WrongFirstTimePrompt] 這個特殊字串,checkForm 一看到就轉去 askForClearPromptFirstTime(),回一段附範例句型的引導文字,讓使用者換個講法。再往下走還有幾個重點:時間不會直接採用,會先丟進 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」,這句話會從頭再跑一次 rewriteLeavePrompt → checkForm,因為代理人、假別、事由都還在 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_search、category_search 負責拿 LLM 判斷出來的字串去跟員工真正可用的清單模糊比對;is_valid_date 則是在事件發生日追問迴圈裡,用來確認使用者回覆的格式對不對。抓到規則後可以照自己公司規定調整細節。
run-gradio-edgeLLM.pyfrom 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 的不確定性優化。