iT邦幫忙

2026 iThome 鐵人賽

DAY 16
0
AI 自動化

AaaS from Scratch: 從一次性定義,到規模化分析系列 第 16 篇

[Day 16] 用 Jev 做 Context Management(1)

  • 分享至 

  • xImage
  •  

上一篇我們用 Jev 做 Skill Routing:

User Request
     ↓
    Jev
     ↓
Relevant Skill
     ↓
   Agent

接下來想處理的是另一個問題:

Conversation 越來越長之後,到底哪些 history 還需要送回 model?

但在開始呼叫 Jev 之前,先建構 conversation history。

現在 Agent 怎麼記住 Conversation?

目前我們使用 LangGraph checkpointer:

self.checkpointer = InMemorySaver()

同一個 thread_id 進來時,Agent 就可以接著前面的 state 繼續執行。

這很好用,但有兩個問題。

第一個是 history 只存在 memory:

Server Restart
→ Conversation Gone

第二個是 Checkpointer 裡的 messages 會一直累積:

User
Assistant
Tool Call
Tool Result
...

每次新的對話出現,這些 history 又會重新送進 model。

在我的一個 test conversation 裡,87% 的 context 都是舊 turn 留下來的 tool traffic,問題不只是 token 變多,而是:

所有的內容都是需要送給 model 的嗎?

Deep Agents 已經有 Context Management?

Deep Agents 本身有 summarization、large tool result eviction 和 subagent 這類機制。

但它們比較像:

不要讓 context 爆掉。

不是:

只留下這次 request relevant 的 context。

也就是:

Deep Agents
→ size / age

接下來想做的
→ relevance

所以我會把原本的 context management 當成 safety net,而不是主要的 filtering mechanism。

Own the History First

如果之後想做到:

Conversation History
        ↓
       Jev
        ↓
 Relevant Turns
        ↓
      Agent

那我們必須先能自己:

store
read
group
rebuild
filter

所以今天先不做 filtering。

先把 conversation history 從 checkpointer 裡拿出來,變成 application 自己管理的資料。

一個 Conversation 一個 JSONL

先在 local 端用檔案來儲存對話紀錄,都設計完才找資料庫儲存:

conversations/<thread_id>.jsonl

一個 message 一行:

{"id":"9f2c...","previous":null,"turn":1,"role":"user","content":"Which category has the most views?"}
{"id":"41ab...","previous":"9f2c...","turn":1,"role":"assistant","tool_calls":[...]}
{"id":"c07e...","previous":"41ab...","turn":1,"role":"tool","content":"[...]"}
{"id":"5d19...","previous":"c07e...","turn":1,"role":"assistant","content":"Music has the most views."}

其中幾個欄位之後會特別有用:

turn
→ filtering 可以以整個 turn 為單位

tool_call_id
→ 保留 tool call 和 result 的關係

previous
→ 搬到 database 後仍然知道順序

為什麼 JSONL?

主要有三個原因:

Append simple
Crash friendly
Easy to move to database

每個 message 出現時直接 append,不需要重寫整份 conversation。

如果 process 中途 crash,最多留下最後一個 incomplete line,前面的 records 還是完整的。

之後如果搬到 database 也很容易處理:

One line
→ One row

這邊有一點要注意:如果 crash 留下一個 incomplete line,下一次 append 不能直接接在後面,不然新的 record 也會一起壞掉。

所以 write 前會先確認 file 最後有沒有 newline。

把 History 變回 Messages

存下來之後,下一步是把 records rebuild 成 Agent 看得懂的 messages:

user
→ HumanMessage

assistant
→ AIMessage

tool
→ ToolMessage

Conceptually:

def to_messages(records):
    messages = []

    for record in records:
        if record["role"] == "user":
            messages.append(HumanMessage(record["content"]))

        elif record["role"] == "assistant":
            messages.append(AIMessage(...))

        elif record["role"] == "tool":
            messages.append(ToolMessage(...))

    return messages

還有一個特殊 case。

如果 Agent 已經產生 tool call,但 execution 還沒回來 process 就掛掉:

Tool Call
✓ stored

Tool Result
✗ missing

rebuild 時會補上一個 error ToolMessage,避免產生 invalid message sequence。

不用 Checkpointer 會不會變差?

把 history 自己 rebuild 之後,我做了一個簡單的 multi-turn test。

裡面有:

"that"
"that category"
"which came second"

這類需要依賴 earlier turns 的 follow-up。

比較:

A
→ LangGraph Checkpointer

B
→ Rebuild from JSONL

結果兩邊都是:

Correct   4/4
Grounded  4/4

至少在這組測試裡,把 history 拿回 application 自己管理後,Agent 還是可以正常理解 follow-up。

當然這不代表兩種方式完全等價。

像 human approval / interrupt 這類功能,還是可能需要 checkpointer。

下一篇會開始思考要怎麼用 Jev 來處理一下問題:

根據這次 query,過去的歷史紀錄中有哪些內容需要被加入context?


上一篇
[Day 15] 用 Jev 來做 Skill Routing
下一篇
[Day 17] 用 Jev 做 Context Management(2)
系列文
AaaS from Scratch: 從一次性定義,到規模化分析 共 17 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言