昨天我們已經完成了最小可跑的版本,連結了查資料庫、組 prompt、請模型回答,我們現在要加上兩個功能也就是工具化(Tool)與記憶(Memory)。
簡單說一下我們要做的事情:
但必須要說這便做的只是我們手動決定流程的部分,並不是自動決策的 Agent ,有了agent_single.py
我們才好去驗證是哪個地方出錯。
這邊我們只會放上新增的項目,不要把以前的資料夾都刪掉噢><
project/
├─ rag/
│ └─ connect.py # 新增「第N條查詢」小工具 + build_prompt 支援記憶
├─ utils/
│ └─ memory.py # 儲存最近 Q&A 摘要
└─ agent/agent_single.py # 自動判斷是不是「第N條」,回答後寫入記憶 & 統一日誌
utils/memory.py
這部分是在儲存最近的對話摘要,每次問完問題,就會寫進 memory/session.json
,下次讀取時再帶進 prompt。
# utils/memory.py
from pathlib import Path
from datetime import datetime
import json
MEM_DIR = Path("memory")
MEM_PATH = MEM_DIR / "session.json"
MAX_ITEMS = 8 # 只留最近 8 筆
def _brief(s: str, n: int = 80) -> str:
s = (s or "").replace("\n", " ").strip()
return s[:n] + ("..." if len(s) > n else "")
def load_session() -> dict:
if not MEM_PATH.exists():
return {"updated_at": None, "items": []}
try:
return json.loads(MEM_PATH.read_text(encoding="utf-8"))
except Exception:
return {"updated_at": None, "items": []}
def save_session(data: dict) -> None:
MEM_DIR.mkdir(parents=True, exist_ok=True)
data["updated_at"] = datetime.now().isoformat(timespec="seconds")
MEM_PATH.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
def remember(question: str, answer: str) -> None:
d = load_session()
items = d.get("items", [])
items.append({
"ts": datetime.now().isoformat(timespec="seconds"),
"q": question,
"a": answer,
"q_brief": _brief(question),
"a_brief": _brief(answer),
})
d["items"] = items[-MAX_ITEMS:]
save_session(d)
def memory_lines() -> list[str]:
d = load_session()
return [f"Q: {i['q_brief']} | A: {i['a_brief']}" for i in d.get("items", [])]
rag/connect.py
這邊加入查詢「第N條」的工具,不想寫得太複雜,就只抓數字就好,它會從問題裡找出條號(例如「第14條」),然後直接從 ChromaDB 撈出該條文內容。。
# 新增這部分就好
import re
def _norm_article_no(q: str) -> str:
"""
從問題中抓出條號,回傳標準字串:例如 '第14條';抓不到回空字串。
(簡化版:只處理阿拉伯數字,先不處理中文數字)
"""
# 多允許「是什麼」或「內容」這種後綴也能抓到
m = re.search(r"第?\s*([0-90-9]{1,3})\s*條", q)
if not m:
# 抓「第14條是什麼」「第14條內容」「第 14 條規定」這種
m = re.search(r"第?\s*([0-90-9]{1,3})\s*條[^。!?]*", q)
if not m:
# 問題裡只要有 1~3 位數字也算條號
m = re.search(r"([0-90-9]{1,3})", q)
if not m:
return ""
num = re.sub(r"[0-9]", lambda d: str(ord(d.group()) - 65248), m.group(1))
return f"第{int(num)}條"
def article_lookup(query_or_no: str) -> str:
"""
抓到 → 回傳「章名|第N條 + 全文(把同條的 chunk 串起來)」;miss → 回空字串。
"""
target = _norm_article_no(query_or_no)
if not target:
return ""
got = coll.get(where={"section_id": {"$eq": target}}, include=["documents","metadatas"])
docs = (got.get("documents") or [])
metas = (got.get("metadatas") or [])
if not docs:
return ""
head = f"{metas[0].get('chapter','')}|{target}" if metas else target
return (head + "\n" + "\n".join(docs)).strip()
為了保存記憶這邊要對 build_prompt 小小的更新一下
import textwrap
from utils.memory import memory_lines # 新增 import
def build_prompt(query, hits, mem_lines: list[str] | None = None):
blocks = []
for i, h in enumerate(hits, 1):
header = f"[{i}] {h.get('chapter','')} | {h.get('section_id','')} | 距離={float(h.get('distance',0.0)):.4f}"
body = f"```text\n{(h.get('text','') or '').replace('\n',' ')}\n```"
blocks.append(header + "\n" + body)
context = "\n\n".join(blocks) if blocks else "(本次查無相關段落)"
mem = "" if not mem_lines else "【最近對話要點】\n" + "\n".join([f"- {m}" for m in mem_lines]) + "\n"
return textwrap.dedent(f"""
你是一位專業的台灣資安法規顧問。若文件沒有明確資訊,請答「文件中沒有相關內容」。
回答請用中文條列式,並在每條末尾標註來源索引。
{mem}【相關文件內容】
{context}
【問題】
{query}
""").strip()
agent/agent_single.py
這邊會先 check 一下有沒有說到「第N條」,有的話我們就會用 article_lookup()
,沒有就繼續用原先的RAG 最後面會用 remember()
記 Q/A
# import 多兩個
from utils.memory import remember, memory_lines
from rag.connect import article_lookup, build_prompt
# main() 裡原本 parse_args / search 流程要保留,
# 把「產生回答」這段改成這樣:
# 先判斷是不是「第N條」
direct = article_lookup(args.q)
if direct:
answer = direct
mode = "tool" # 用到條文直接查詢工具
else:
hits = search_chunks(args.q, k=args.k)
pretty_print_hits(hits)
prompt = build_prompt(args.q, hits, mem_lines=memory_lines()) # ← 加入記憶
answer = ask_ollama(prompt, model=args.model).strip()
mode = "rag"
print("\n=== 最終回答 ===")
print(answer)
# 記憶:寫入這次問答
remember(args.q, answer)
# 統一 JSONL 日誌(多一個 mode 欄位)
if args.json_out:
from datetime import datetime
from pathlib import Path
import json
logs_dir = Path("logs"); logs_dir.mkdir(parents=True, exist_ok=True)
outpath = logs_dir / f"run_{datetime.now():%Y%m%d}.jsonl"
record = {
"ts": datetime.now().isoformat(timespec="seconds"),
"mode": mode, # "tool" | "rag"
"query": args.q,
"k": args.k,
"model": args.model,
"hits": hits if mode == "rag" else [],
"answer": answer,
}
outpath.open("a", encoding="utf-8").write(json.dumps(record, ensure_ascii=False) + "\n")
print(f"\n[log] 已寫入:{outpath}")
最後可以小小測試一下:
python agent/agent_single.py --q "第1條是什麼?" --json_out
這邊都是為了日後讓 Agent 能夠用工具查條文,也能記得你剛問過什麼做鋪成,還沒真正開始。
明天我們會在這個基礎上,正式導入 AutoGen,讓模型自己決定該用哪個工具,變得更自動、更像一個真正的「智慧助理」。