iT邦幫忙

2025 iThome 鐵人賽

DAY 26
0
AI & Data

RAG × Agent:從知識檢索到智慧應用的30天挑戰系列 第 26

Day 26|實戰工具化與記憶:讓系統會查條文、會記得你說過什麼

  • 分享至 

  • xImage
  •  

昨天我們已經完成了最小可跑的版本,連結了查資料庫、組 prompt、請模型回答,我們現在要加上兩個功能也就是工具化(Tool)與記憶(Memory)。
簡單說一下我們要做的事情:

  • 工具:我們會加上一個小功能,也就是如果使用者問「第幾條是什麼」,系統就會自動查出整條法條,而不是去跑整個 RAG
  • 記憶:讓 Agent 能記得最近問過什麼,下次問相關內容時可以把之前問過的內容摘要放入 prompt

但必須要說這便做的只是我們手動決定流程的部分,並不是自動決策的 Agent ,有了agent_single.py我們才好去驗證是哪個地方出錯。


檔案架構

這邊我們只會放上新增的項目,不要把以前的資料夾都刪掉噢><

project/
├─ rag/
│  └─ connect.py        # 新增「第N條查詢」小工具 + build_prompt 支援記憶
├─ utils/
│  └─ memory.py         # 儲存最近 Q&A 摘要
└─ agent/agent_single.py     # 自動判斷是不是「第N條」,回答後寫入記憶 & 統一日誌

1. 建立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", [])]

2. 修改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()

3. 修改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,讓模型自己決定該用哪個工具,變得更自動、更像一個真正的「智慧助理」。


上一篇
Day 25|實戰 Agent 入口設計:從查詢到答案(RAG×ChromaDB×Ollama)
系列文
RAG × Agent:從知識檢索到智慧應用的30天挑戰26
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言