1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
| import sys
import os
import re
import urllib.parse
import json
from datetime import datetime, timedelta
from pprint import pprint
# 統一從 typing 導入，並移除冗餘導入
from typing import Literal, List, Dict, Any, Optional
# LangChain Core/Community Imports
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import SQLDatabaseToolkit, create_sql_agent
from langchain_core.prompts import PromptTemplate
# 統一使用 Pydantic v2 標準
from pydantic import BaseModel, Field, ValidationError
from langchain_core.output_parsers import PydanticOutputParser
# OpenAI LLM
from langchain_openai import ChatOpenAI # 建議在環境變數設置 OPENAI_API_KEY
# FastAPI / LINE Bot Imports
from fastapi import FastAPI, Request, HTTPException
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage
# 查詢結果的單筆資料結構 (Query data list item)
class ExpenseData(BaseModel):
Category: str = Field(description="支出的具體類別。")
Amount: float = Field(description="支出金額。")
Description: str = Field(description="支出事項的簡短描述。")
CreatedAt: str = Field(description="完整日期時間字串。")
UserId : Optional[str] = Field(description="使用者編號")
# 修正後的通用最終回覆結構 (Final Answer Schema)
# 必須使用 Optional/預設值處理 insert 和 query 之間的互斥欄位
class FinalResponse(BaseModel):
# 共同欄位
action: Literal["insert", "query"] = Field(description="要執行的動作，只能是 'insert' 或 'query'。")
message: str = Field(description="給長輩的嘮叨或提醒訊息，必須是中文長字串，每次都必須不同。")
# INSERT 專屬欄位 (QUERY 時為 None/空字串/0.0)
category: Optional[str] = Field(None, description="支出的具體類別。")
amount: Optional[float] = Field(None, description="支出金額。")
description: Optional[str] = Field(None, description="支出事項的簡短描述。")
created_at: Optional[str] = Field(None, description="完整日期時間字串。")
user_id : Optional[str] = Field(None, description="使用者編號")
# QUERY 專屬欄位 (INSERT 時為 None/空列表/空字串)
sql: Optional[str] = Field(None, description="Agent 執行過的 SQL SELECT 語句。")
data: Optional[List[ExpenseData]] = Field(None, description="SQL 查詢回傳的資料列表。")
app = FastAPI(title="長輩版中文記帳助理")
# sql server
params = urllib.parse.quote_plus(
"DRIVER={ODBC Driver 17 for SQL Server};"
"SERVER=(localdb)\\MSSQLLocalDB;"
"DATABASE=ExpenseDB;"
"Trusted_Connection=yes;"
)
connection_str = f"mssql+pyodbc:///?odbc_connect={params}"
try:
db = SQLDatabase.from_uri(connection_str)
except Exception as e:
print(f"🚨 資料庫連線錯誤：{e}")
sys.exit(1)
# llm parser
llm = ChatOpenAI(temperature=0, model="gpt-4o")
# llm = ChatOpenAI(temperature=0)
# PydanticOutputParser 使用單一的 FinalResponse 模型
pydantic_parser = PydanticOutputParser(pydantic_object=FinalResponse)
format_instructions = pydantic_parser.get_format_instructions()
# sql agent
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
agent_executor = create_sql_agent(
llm=llm,
toolkit=toolkit,
agent_type="openai-tools",
verbose=True,
agent_executor_kwargs={"return_intermediate_steps": True},
)
prompt = PromptTemplate(
input_variables=["user_input", "today", "user_id", "format_instructions"],
template=(
"你是一位親切但愛嘮叨的台灣長輩記帳助理，會用長輩語氣提醒使用者花費要節省。\n\n"
"資料表結構：[Expenses(UserId NVARCHAR(100), Category NVARCHAR(50), Amount DECIMAL(10,2), Description NVARCHAR(200), CreatedAt DATETIME, UserId NVARCHAR(250))]\n"
"類別(Category) 分為：食、衣、住、行、育、樂、其他。\n"
"CreatedAt 欄位必須為完整日期時間格式，例如 '2025-10-25 10:30:00.000'\n\n"
"【規則】\n"
" 當使用者要求新增支出，生成正確的 SQL INSERT 語句，並在資料中包含 UserId。\n"
" 確保字串用單引號，數字型別正確。\n"
" 執行後回覆 JSON，例如：\n"
"{{'action':'insert','user_id':'{user_id}','category':'...','amount':...,'description':'...','created_at':'...','message':'隨機生成長輩嘮叨語氣，每次都不同'}}\n\n"
" 💡 message 範例可使用以下長輩語氣，但請每次隨機挑選或創造相似語氣，不要固定使用同一句：\n"
" - '唉呀～又花錢啦？要省一點哪～'\n"
" - '年輕人要懂理財啊～今天花太多了喔～'\n"
" - '唉唷～這筆支出有點大，記得下個月要省錢啊～'\n"
" - '哎呀～又出去吃飯啦？要學會自己煮飯省點錢啊～'\n"
" - '你真的很討債～今天不要亂花錢啊～'\n"
" - '歸剛ㄟ～買東西要有計畫，別亂花錢～'\n"
" - '唉唷～小朋友啊，這筆花費有點大，先忍一忍吧～'\n"
" - '年輕人要存錢啦～吃喝玩樂要適可而止～'\n"
"請生成符合規則的 JSON，message 必須使用長輩嘮叨語氣，每次都不同。\n\n"
" 當使用者要求查詢或統計，生成正確 SQL SELECT，並回覆 JSON，例如：\n"
"{{'action':'query','user_id':'{user_id}','sql':'SELECT ...','message':'你的花費','data':[{{'Category':'食','Amount':50.0,'Description':'買飯','CreatedAt':'2025-10-24 12:00:00'}}]}}\n\n"
"回覆 **只包含 JSON**，不要解釋或重複資料表結構。\n"
"如果輸入含時間詞（今天、昨天、前天等），請轉換成完整日期時間格式。\n\n"
"今天日期：{today}\n"
"使用者 ID：{user_id}\n"
"使用者輸入：{user_input}\n\n"
"請回覆 **JSON**，格式如下：\n"
"{format_instructions}"
)
)
# 轉換時間
def replace_time_words(user_input: str):
now = datetime.now()
found_date = False
# 基本日期詞
mapping = {"今天": now, "昨天": now - timedelta(days=1), "前天": now - timedelta(days=2)}
for word, dt in mapping.items():
if re.search(r'\b' + re.escape(word) + r'\b', user_input):
dt_str = dt.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
user_input = re.sub(r'\b' + re.escape(word) + r'\b', dt_str, user_input)
found_date = True
# 星期詞
weekday_mapping = {
"禮拜一": 0, "禮拜二": 1, "禮拜三": 2, "禮拜四": 3,
"禮拜五": 4, "禮拜六": 5, "禮拜日": 6,
"星期一": 0, "星期二": 1, "星期三": 2, "星期四": 3,
"星期五": 4, "星期六": 5, "星期日": 6,
"週一": 0, "週二": 1, "週三": 2, "週四": 3,
"週五": 4, "週六": 5, "週日": 6,
}
for word, target_weekday in weekday_mapping.items():
pattern = rf"(上週|本週)?{word}"
for match in re.finditer(pattern, user_input):
prefix = match.group(1)
today_weekday = now.weekday()
delta_days = target_weekday - today_weekday
if prefix == "上週":
delta_days -= 7
elif prefix == "本週" or prefix is None:
if delta_days > 0:
delta_days -= 7
target_date = now + timedelta(days=delta_days)
dt_str = target_date.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
user_input = user_input.replace(match.group(0), dt_str)
found_date = True
# 本週/上週/這個月/上個月
week_month_mapping = {
"這週": now - timedelta(days=now.weekday()),
"上週": now - timedelta(days=now.weekday() + 7),
"這個月": now.replace(day=1),
"上個月": (now.replace(day=1) - timedelta(days=1)).replace(day=1),
}
for word, dt in week_month_mapping.items():
if re.search(r'\b' + re.escape(word) + r'\b', user_input):
dt_str = dt.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
user_input = re.sub(r'\b' + re.escape(word) + r'\b', dt_str, user_input)
found_date = True
# 處理時間詞
match = re.search(r'(\d+)\s*點', user_input)
if match:
hour = int(match.group(1))
date_match = re.search(r'\d{4}-\d{2}-\d{2}', user_input)
if date_match:
user_input = re.sub(r'(\d+)\s*點', '', user_input)
user_input = user_input.strip() + f" {hour:02d}:00:00.000"
found_date = True
# 沒有日期時間，預設加今天
if not found_date and not re.search(r'\d{4}-\d{2}-\d{2}', user_input):
user_input += f" {now.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]}"
return user_input
CHANNEL_ACCESS_TOKEN = os.getenv("CHANNEL_ACCESS_TOKEN")
# print(CHANNEL_ACCESS_TOKEN)
CHANNEL_SECRET = os.getenv("CHANNEL_SECRET")
# print(CHANNEL_SECRET)
line_bot_api = LineBotApi(CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(CHANNEL_SECRET)
# line callback
@app.post("/callback")
async def callback(request: Request):
signature = request.headers.get("X-Line-Signature", "")
body = await request.body()
body_str = body.decode("utf-8")
try:
handler.handle(body_str, signature)
except InvalidSignatureError:
raise HTTPException(status_code=400, detail="Invalid signature")
return "OK"
# 當使用者發送訊息觸發
@handler.add(MessageEvent, message=TextMessage)
def handle_message(event: MessageEvent):
user_id = event.source.user_id
user_text = event.message.text
print(f"收到來自使用者 ID: {user_id} 的訊息: {user_text}")
# 預設訊息
reply_message = "阿公聽不懂你在說什麼啦～"
try:
# 處理時間詞
user_input = replace_time_words(user_text)
# 這裡使用當前的時間作為 TODAY
today = datetime.now().strftime("%Y-%m-%d")
# 準備 final prompt (新增 user_id 傳入)
final_prompt = prompt.format(
user_input=user_input,
today=today,
user_id=user_id,
format_instructions=pydantic_parser.get_format_instructions()
)
# 呼叫 agent
response = agent_executor.invoke({"input": final_prompt})
output_text = response.get("output", "")
pprint(output_text)
# 如果 Agent 輸出格式錯誤，會拋出異常並被 except 捕獲
parsed_data: FinalResponse = pydantic_parser.parse(output_text)
action = parsed_data.action
# 根據 action 選擇對應邏輯 (還原 DB 寫入邏輯)
if action == "insert":
# 確保所有必填欄位存在 (Pydantic 已經做了大部分檢查)
if not all([parsed_data.category, parsed_data.amount, parsed_data.description, parsed_data.created_at]):
raise ValueError("解析到的 INSERT 資料不完整。")
# 構建 SQL 語句 (必須包含 UserId 欄位)
sql_to_run = f"""
INSERT INTO Expenses (Category, Amount, Description, CreatedAt, UserId)
VALUES ('{parsed_data.category}', {parsed_data.amount}, '{parsed_data.description}', '{parsed_data.created_at}', '{user_id}');
"""
# 執行資料庫寫入操作
db.run(sql_to_run)
# 回覆 LLM 生成的長輩嘮叨訊息
reply_message = parsed_data.message
elif action == "query":
# 這裡假設 Agent 已經執行了 SELECT SQL 並將結果填入 data
data: List[ExpenseData] = parsed_data.data if parsed_data.data else []
if data:
reply_message = "你的花費明細如下：\n" + "\n".join(
[f"{d.CreatedAt} {d.Category} {d.Amount}元 {d.Description}" for d in data]
)
else:
reply_message = "阿公查不到你的花費資料～"
else:
# 處理無法識別的 action 或其他回覆
reply_message = parsed_data.message
except Exception as e:
reply_message = f"阿公嘆氣：網路線斷掉了，資料解析失敗。({type(e).__name__})"
print(f"致命錯誤 (可能是 JSON 或 Pydantic 解析失敗)：{e}")
# 回覆 LINE
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text=reply_message)
)
|