Day 4 我們設計了 TraceStep 和 AgentTrace,讓 Agent 在執行時可以記錄每一步發生了什麼。
目前流程大概是:
User Task
-> Agent Runner
-> LLM Client
-> Tool Call
-> Tool Result
-> Final Answer
-> 印出 Trace JSON
這已經比只看 final answer 好很多。
但 Day 4 的 trace 還有一個明顯限制:
Trace 只存在記憶體和終端機輸出中,程式結束後就消失了。
如果之後要做 Trace Viewer、Eval Runner、Failure Analysis,就不能只把 trace 印出來,而是要把每次 Agent 執行結果保存下來。
所以 Day 5 要做的事情是:
把 Agent trace 寫進 SQLite,讓每次執行都能被查詢與回放。
今天要建立最小可用的 trace storage。
會完成:
sessions table。steps table。save_trace()。app.py,讓每次 Agent 執行後自動儲存 trace。今天先不做:
這些留到後面幾天。
這個系列要做的是輕量級 Agent 測試驗證平台,不是一開始就做 production 系統。
所以 Day 5 先使用 SQLite。
SQLite 的優點是:
.db 檔案。sqlite3 module,不需要額外套件。以目前需求來說,我們只需要保存:
SQLite 已經足夠。
延續 Day 4,今天新增 storage/ 資料夾。
agent-testing-platform/
app.py
agents/
__init__.py
simple_agent.py
prompts.py
fake_llm.py
tools/
__init__.py
calculator.py
tracing/
__init__.py
models.py
storage/
__init__.py
database.py
schema.sql
data/
agent_traces.db
新增:
| 檔案 | 用途 |
|---|---|
storage/__init__.py |
讓 storage 成為 Python package |
storage/schema.sql |
定義 SQLite table |
storage/database.py |
負責初始化 DB、儲存 trace、查詢 session |
修改:
| 檔案 | 修改內容 |
|---|---|
app.py |
Agent 執行完成後呼叫 save_trace() |
data/agent_traces.db 不需要手動建立,之後程式會自動產生。
今天先設計兩張表:
sessions
steps
sessions 表示一次 Agent 執行。
steps 表示該次執行中的每一個 trace step。
兩者關係是:
sessions 1 -> many steps
也就是一個 session 會有多個 steps。
新增 storage/schema.sql:
CREATE TABLE IF NOT EXISTS sessions (
session_id TEXT PRIMARY KEY,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS steps (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
step_order INTEGER NOT NULL,
step_type TEXT NOT NULL,
name TEXT NOT NULL,
input_data TEXT NOT NULL,
output_data TEXT NOT NULL,
error TEXT,
timestamp TEXT NOT NULL,
FOREIGN KEY (session_id) REFERENCES sessions(session_id)
);
這裡的 sessions.session_id 會對應 Day 4 的 AgentTrace.session_id。
steps.session_id 則用來表示這個 step 屬於哪一次執行。
幾個欄位需要特別說明:
| 欄位 | 說明 |
|---|---|
step_order |
step 在這次 session 中的順序 |
step_type |
例如 user_input、llm_response、tool_call |
input_data |
這一步的輸入資料,會以 JSON 字串儲存 |
output_data |
這一步的輸出資料,會以 JSON 字串儲存 |
error |
如果這一步有錯誤,記錄錯誤訊息 |
SQLite 沒有原生 JSON 欄位型別,所以今天先把 input_data 和 output_data 用 JSON 字串存起來。
新增 storage/__init__.py:
這個檔案今天可以先留空,不需要放任何程式碼。
它的用途是讓 Python 把 storage 資料夾視為 package,這樣 app.py 才能匯入:
from storage.database import init_db, save_trace
新增 storage/database.py:
import json
import sqlite3
from datetime import datetime
from pathlib import Path
from tracing.models import AgentTrace
BASE_DIR = Path(__file__).resolve().parent.parent
DATA_DIR = BASE_DIR / "data"
DB_PATH = DATA_DIR / "agent_traces.db"
SCHEMA_PATH = BASE_DIR / "storage" / "schema.sql"
def get_connection() -> sqlite3.Connection:
DATA_DIR.mkdir(exist_ok=True)
return sqlite3.connect(DB_PATH)
def init_db() -> None:
with get_connection() as conn:
schema = SCHEMA_PATH.read_text(encoding="utf-8")
conn.executescript(schema)
def save_trace(trace: AgentTrace) -> None:
with get_connection() as conn:
conn.execute(
"""
INSERT OR IGNORE INTO sessions (session_id, created_at)
VALUES (?, ?)
""",
(trace.session_id, datetime.now().isoformat()),
)
for index, step in enumerate(trace.steps):
conn.execute(
"""
INSERT INTO steps (
session_id,
step_order,
step_type,
name,
input_data,
output_data,
error,
timestamp
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
trace.session_id,
index,
step.step_type,
step.name,
json.dumps(step.input_data, ensure_ascii=False),
json.dumps(step.output_data, ensure_ascii=False),
step.error,
step.timestamp,
),
)
def list_sessions() -> list[dict]:
with get_connection() as conn:
conn.row_factory = sqlite3.Row
rows = conn.execute(
"""
SELECT
sessions.session_id,
sessions.created_at,
COUNT(steps.id) AS step_count
FROM sessions
LEFT JOIN steps ON sessions.session_id = steps.session_id
GROUP BY sessions.session_id
ORDER BY sessions.created_at DESC
"""
).fetchall()
return [dict(row) for row in rows]
這個檔案做了幾件事。
第一,定義資料庫位置:
DB_PATH = DATA_DIR / "agent_traces.db"
之後 trace 會被存在:
data/agent_traces.db
第二,init_db() 會讀取 storage/schema.sql,並建立需要的 table。
第三,save_trace(trace) 會把一個 AgentTrace 存入資料庫。
它會先寫入 sessions,再把 trace.steps 一筆一筆寫入 steps。
第四,list_sessions() 會查詢目前儲存過的 session,並回傳每個 session 有幾個 step。
這個 function 今天先用來在終端機確認資料有被存進去。Day 6 做 Trace Viewer 時也會用到它。
接著修改 app.py:
import json
from agents.fake_llm import FakeLLMClient
from agents.simple_agent import SimpleAgent
from storage.database import init_db, list_sessions, save_trace
def main():
init_db()
agent = SimpleAgent(llm_client=FakeLLMClient())
user_task = input("Task: ")
result = agent.run(user_task)
save_trace(result.trace)
print("\nAnswer:")
print(result.answer)
if result.tool_calls:
print("\nTool Calls:")
for tool_call in result.tool_calls:
print(f"- tool_name: {tool_call.tool_name}")
print(f" tool_input: {tool_call.tool_input}")
print(f" tool_output: {tool_call.tool_output}")
print("\nTrace:")
print(json.dumps(result.trace.to_dict(), ensure_ascii=False, indent=2))
print("\nSaved Sessions:")
for session in list_sessions():
print(
f"- {session['session_id']} "
f"created_at={session['created_at']} "
f"steps={session['step_count']}"
)
if __name__ == "__main__":
main()
這次 app.py 多了三個重點。
第一,在程式開始時呼叫:
init_db()
這會確保 SQLite database 和 table 已經建立。
第二,在 Agent 執行完成後呼叫:
save_trace(result.trace)
這會把 Day 4 產生的 trace 寫進 SQLite。
第三,最後呼叫:
list_sessions()
用來確認目前資料庫裡有哪些歷史 session。
在專案根目錄執行:
python3 app.py
輸入:
請計算 135 * 28
預期會看到類似結果:
Task: 請計算 135 * 28
Answer:
The result is 3780
Tool Calls:
- tool_name: calculator
tool_input: 135 * 28
tool_output: 3780
Trace:
{
"session_id": "8a2b2f21-d1a4-4d8d-9a57-0db43a4e9d23",
"steps": [
...
]
}
Saved Sessions:
- 8a2b2f21-d1a4-4d8d-9a57-0db43a4e9d23 created_at=2026-09-05T12:00:00.000000 steps=5
如果你重複執行幾次,Saved Sessions 應該會出現多筆資料。
看到這段輸出,就表示 trace 不再只是印在終端機,而是真的被保存到 SQLite。
如果你的電腦有 sqlite3 指令,可以在專案根目錄執行:
sqlite3 data/agent_traces.db "SELECT session_id, created_at FROM sessions;"
也可以查詢 steps:
sqlite3 data/agent_traces.db "SELECT step_order, step_type, name FROM steps ORDER BY id;"
可能會看到:
0|user_input|User Task
1|llm_response|LLM Response
2|tool_call|Tool Call
3|tool_result|Tool Result
4|final_answer|Final Answer
這就是一次 Agent 執行被拆成多個 step 後,存進資料庫的結果。
如果你的環境沒有 sqlite3 指令也沒關係,app.py 裡的 list_sessions() 已經可以確認資料有被寫入。
這裡補充一件實作時容易遇到的事情。
如果你故意輸入:
請計算 135 28
目前程式仍然可能直接拋出:
ValueError: No arithmetic expression found
這不是最終平台的理想狀態。
理想上,Agent 應該把這類錯誤記錄成 trace step,並回傳一個可讀的錯誤結果,而不是直接中斷。
但依照目前 30 天規劃,Day 5 的重點是:
把正常執行產生的 trace 存進 SQLite。
錯誤捕捉、失敗分類、failure type 會留到第三週的 Failure Analysis 階段處理。
也就是說,今天先完成「正常流程的持久化」,後面再補上「錯誤流程的持久化」。
其實可以。
最簡單的設計是只建立一張表:
traces
-> session_id
-> trace_json
這樣實作最快,但後面查詢會比較不方便。
例如之後會想問:
tool_call?如果整包 trace 都存在一個 JSON 欄位裡,這些查詢會比較麻煩。
所以今天採用比較結構化的設計:
sessions
steps
這樣 Day 6 做 Trace Viewer,或後面做 Failure Dashboard,都會比較容易。
今天完成後,系統具備:
sessions table。steps table。save_trace()。list_sessions()。目前還沒有:
Day 4 讓 Agent 可以產生 trace。
Day 5 則讓 trace 可以被保存。
這一步很重要。只要 trace 能存下來,後面就可以做:
今天的資料庫設計先保持簡單:
sessions
-> session_id
-> created_at
steps
-> session_id
-> step_order
-> step_type
-> name
-> input_data
-> output_data
-> error
-> timestamp
這個設計不複雜,但已經足夠支撐接下來的 Trace Viewer。
Day 6 會建立簡單 Trace Viewer。
目前已經能把 trace 存進 SQLite,但查看方式還很陽春,只能透過終端機或 SQL 查詢。
下一篇會使用 Streamlit 做一個簡單介面,讓我們可以:
到那時候,Agent 的執行過程就會從「資料庫裡的紀錄」變成「看得懂的 Trace Viewer」。