iT邦幫忙

2026 iThome 鐵人賽

DAY 5
0

前言

Day 4 我們設計了 TraceStepAgentTrace,讓 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。

會完成:

  1. 建立 SQLite database。
  2. 建立 sessions table。
  3. 建立 steps table。
  4. 實作 save_trace()
  5. 實作查詢歷史 session 的 function。
  6. 修改 app.py,讓每次 Agent 執行後自動儲存 trace。

今天先不做:

  • Streamlit Trace Viewer。
  • Dashboard。
  • Evaluation result storage。
  • Failure classification。

這些留到後面幾天。


為什麼選 SQLite?

這個系列要做的是輕量級 Agent 測試驗證平台,不是一開始就做 production 系統。

所以 Day 5 先使用 SQLite。

SQLite 的優點是:

  • 不需要額外安裝資料庫服務。
  • 資料會存在單一 .db 檔案。
  • 很適合小型工具、prototype 和本機實驗。
  • Python 內建 sqlite3 module,不需要額外套件。

以目前需求來說,我們只需要保存:

  • 一次 Agent 執行,也就是 session。
  • session 裡面的每個 trace step。

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。


建立 schema.sql

新增 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_inputllm_responsetool_call
input_data 這一步的輸入資料,會以 JSON 字串儲存
output_data 這一步的輸出資料,會以 JSON 字串儲存
error 如果這一步有錯誤,記錄錯誤訊息

SQLite 沒有原生 JSON 欄位型別,所以今天先把 input_dataoutput_data 用 JSON 字串存起來。


建立 storage package

新增 storage/__init__.py

這個檔案今天可以先留空,不需要放任何程式碼。

它的用途是讓 Python 把 storage 資料夾視為 package,這樣 app.py 才能匯入:

from storage.database import init_db, save_trace

實作 database.py

新增 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:執行後儲存 trace

接著修改 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 指令,可以在專案根目錄執行:

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 階段處理。

也就是說,今天先完成「正常流程的持久化」,後面再補上「錯誤流程的持久化」。


為什麼不直接把整包 trace JSON 存成一欄?

其實可以。

最簡單的設計是只建立一張表:

traces
  -> session_id
  -> trace_json

這樣實作最快,但後面查詢會比較不方便。

例如之後會想問:

  • 哪些 session 有 tool_call
  • 哪些 step 發生錯誤?
  • 平均每次執行有幾個 step?
  • 哪些 tool 最常被呼叫?

如果整包 trace 都存在一個 JSON 欄位裡,這些查詢會比較麻煩。

所以今天採用比較結構化的設計:

sessions
steps

這樣 Day 6 做 Trace Viewer,或後面做 Failure Dashboard,都會比較容易。


今天完成後的系統狀態

今天完成後,系統具備:

  • SQLite database。
  • sessions table。
  • steps table。
  • save_trace()
  • list_sessions()
  • 每次 Agent 正常執行後會保存 trace。
  • 可以查詢歷史 session。

目前還沒有:

  • Streamlit Trace Viewer。
  • 圖形化 timeline。
  • 錯誤流程完整保存。
  • Evaluation dataset。
  • Failure classification。

今天的重點整理

Day 4 讓 Agent 可以產生 trace。

Day 5 則讓 trace 可以被保存。

這一步很重要。只要 trace 能存下來,後面就可以做:

  • 查詢歷史執行紀錄。
  • 建立 Trace Viewer。
  • 對照測試結果與執行步驟。
  • 分析工具呼叫行為。
  • 統計錯誤發生在哪些 step。

今天的資料庫設計先保持簡單:

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 做一個簡單介面,讓我們可以:

  • 查看 session list。
  • 選擇某一次 Agent 執行。
  • 顯示 step timeline。
  • 查看 tool call 與 final answer。
  • 顯示錯誤訊息。

到那時候,Agent 的執行過程就會從「資料庫裡的紀錄」變成「看得懂的 Trace Viewer」。


上一篇
Day 4:設計 Trace:記錄 Agent 每一步
下一篇
Day 6|建立簡單 Trace Viewer
系列文
從黑盒到可驗證:30 天打造 AI Agent 的 Trace、Eval 與 Guardrails 系統8
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言