Pydantic AI 對話歷史與訊息管理

大型語言模型 (LLM) 本質上是「無狀態 (Stateless)」的。這意味著當你向模型發送一個新問題時，模型本身並不記得你們上一句聊了什麼。為了讓 AI 代理人具備「記憶」，能夠進行上下文連貫的多輪對話，開發者必須在每次發送請求時，將過去的「對話歷史 (Message History)」一併傳送給模型。

Pydantic AI 提供了優雅且直覺的介面，幫助你管理、儲存以及動態處理這些對話歷史。

取得對話結果中的訊息

每次你執行 Agent，Agent 都會與模型進行多次訊息交換（包含使用者的提示、工具的呼叫、模型的內部思考以及最終的回覆）。Pydantic AI 的執行結果（無論是同步的 RunResult 還是串流的 StreamedRunResult）都包含了這些訊息的完整記錄。

你可以透過以下方法來存取這些訊息：

• result.new_messages()：回傳本次執行所產生的所有新訊息列表。這通常包含你剛發出的請求以及模型的最新回覆與工具呼叫過程。
• result.all_messages()：回傳完整的對話歷史列表，包含本次產生的新訊息，以及你之前傳入的歷史訊息。

此外，如果你需要直接取得 JSON 格式的二進位資料（通常用於儲存或網路傳輸），可以使用 new_messages_json() 與 all_messages_json() 兩個變體方法。

傳遞對話歷史以延續對話

要建立一個多輪對話的 Chatbot，關鍵在於將上一次執行產生的 all_messages() 作為下一次執行的輸入參數 message_history 傳入。

如果 message_history 參數有值且不為空，系統就不會自動產生新的 System Prompt，因為它假設你傳入的歷史訊息中已經包含了 System Prompt。如果你的歷史記錄來源（例如從資料庫讀回）遺失了 System Prompt，可以搭配 ReinjectSystemPrompt 功能強制重新注入。

以下是實作連續對話的基礎範例：

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o-mini', instructions='你是一個幽默的助手。')

# 第一輪對話，不需傳入 message_history
result1 = agent.run_sync('請告訴我一個關於程式設計師的笑話。')
print(f"AI: {result1.data}")

# 第二輪對話，傳入第一輪的完整歷史 (all_messages)
result2 = agent.run_sync(
    '可以解釋一下這個笑話的笑點嗎？',
    message_history=result1.all_messages()
)
print(f"AI: {result2.data}")

對話追蹤與 Conversation ID

在複雜的應用程式中，我們經常需要追蹤或除錯特定的多輪對話。Pydantic AI 的每一個訊息都會帶有兩個重要的識別碼：

• run_id：針對單次 Agent 執行的唯一識別碼（對應到 OpenTelemetry 中的 gen_ai.agent.call.id）。
• conversation_id：跨越多次執行、共用同一個 message_history 的對話識別碼（對應到 gen_ai.conversation.id）。

系統會在第一次執行時自動生成一個 conversation_id，並在後續透過 message_history 傳遞時自動繼承。這讓你可以非常輕鬆地在 Logfire 等觀測工具中將多輪對話關聯起來。

如果你需要從現有的對話分支（Fork）出一個新對話，或者想使用應用程式自訂的資料庫 ID，可以透過 conversation_id 參數進行覆寫：

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o-mini')
result1 = agent.run_sync('請給我一個冷笑話。')

# 預設情況下，如果沒有指定 conversation_id，它會從 message_history 中繼承
result2 = agent.run_sync('再給我一個。', message_history=result1.all_messages())
assert result1.conversation_id == result2.conversation_id

# 使用 'new' 來強制分支，產生一個全新的 conversation_id
forked_result = agent.run_sync(
    '換個話題吧。', 
    message_history=result1.all_messages(), 
    conversation_id='new'
)
assert forked_result.conversation_id != result1.conversation_id

# 或者直接指定你自訂的 ID
custom_result = agent.run_sync('繼續。', conversation_id='my-custom-thread-123')

訊息的儲存與載入 (JSON 序列化)

雖然在記憶體中維護對話狀態對於簡單腳本來說很足夠，但在實際的 Web 應用中，你通常需要將對話歷史存入資料庫，並在下一次 HTTP 請求來臨時讀取出來。

Pydantic AI 提供了 ModelMessagesTypeAdapter，專門用來處理結構化訊息的 JSON 序列化與反序列化。

from pydantic_core import to_jsonable_python, to_json
from pydantic_ai import Agent, ModelMessagesTypeAdapter

agent = Agent('openai:gpt-4o-mini')

# 1. 執行並取得歷史
result1 = agent.run_sync('哈囉！')
history_step_1 = result1.all_messages()

# 2. 將訊息序列化為 JSON 格式（可以直接存入資料庫的 JSON 欄位）
# to_jsonable_python 會將資料轉換為基礎的 Python 字典與列表
jsonable_data = to_jsonable_python(history_step_1)

# 若需要字串格式，可使用 to_json()
# json_string = to_json(history_step_1)

# --- 假設此處經過了資料庫存取與讀取 ---

# 3. 將 JSON 資料反序列化回 Pydantic AI 可以看懂的結構化物件
restored_history = ModelMessagesTypeAdapter.validate_python(jsonable_data)
# 若從字串還原：ModelMessagesTypeAdapter.validate_json(json_string)

# 4. 帶著還原後的歷史繼續對話
result2 = agent.run_sync('我剛剛說了什麼？', message_history=restored_history)
print(result2.data)

訊息的結構 (Message Types)

Pydantic AI 的對話歷史並非單純的字串陣列，而是由 ModelMessage 構成的結構化物件陣列。主要的訊息型別分為兩大類：

向模型發出的請求 (ModelRequest)

代表從應用程式端傳送給模型的訊息。它內部包含多個段落 (Part)，常見的有：

• SystemPromptPart：系統提示，設定模型的人設或規則。
• UserPromptPart：使用者的實際輸入。
• ToolReturnPart：模型呼叫工具後，應用程式執行完畢並回傳給模型的結果。
• RetryPromptPart：模型輸出格式錯誤時，要求重新嘗試的提示。

模型的回覆 (ModelResponse)

代表模型運算後回傳給應用程式的結果。同樣包含多個段落 (Part)：

• TextPart：模型回覆的純文字解答。
• ToolCallPart：模型決定呼叫某個工具的指令。

透過這些結構化物件，你可以預先載入對話歷史（Pre-loading Context），手動偽造出一段「先前的對話」，這在實作 Few-Shot Prompting（少樣本提示）時非常實用：

from pydantic_ai import Agent
from pydantic_ai.messages import ModelRequest, ModelResponse, UserPromptPart, TextPart

agent = Agent('openai:gpt-4o-mini')

# 手動建立結構化對話歷史，作為範例樣本
few_shot_history = [
    ModelRequest(parts=[UserPromptPart(content='巴黎的鐵塔叫什麼？')]),
    ModelResponse(parts=[TextPart(content='艾菲爾鐵塔 (Eiffel Tower)')])
]

# 帶著預設的歷史繼續執行
result = agent.run_sync('那東京的呢？', message_history=few_shot_history)
print(result.data)

動態處理與過濾歷史訊息 (History Processors)

隨著對話輪數增加，歷史訊息會越來越長，不僅耗費 Token 成本，也可能超過模型的上下文長度上限。Pydantic AI 提供了一個強大的機制：History Processors。

你可以在建立 Agent 時透過 history_processors 參數傳入多個處理函式，這些函式會在訊息被送往模型「之前」進行攔截與修改。常見的應用場景包含：過濾無用訊息、保留最近幾筆對話，或是將過舊的對話總結成摘要。

範例：只保留最近的訊息

以下範例展示了如何透過 History Processor，在每次送出請求前，自動裁減對話歷史，只保留最近的五筆記錄。

from pydantic_ai import Agent, ModelMessage

# 定義一個非同步的 history processor
async def keep_recent_messages(messages: list[ModelMessage]) -> list[ModelMessage]:
    """為節省 Token，只保留最後 5 筆訊息"""
    if len(messages) > 5:
        return messages[-5:]
    return messages

# 將處理器綁定到 Agent 上
agent = Agent(
    'openai:gpt-4o-mini', 
    history_processors=[keep_recent_messages]
)

# 當你傳入非常長的歷史時，處理器會自動將其截斷再送給 LLM
long_history = [...] # 假設這裡有一長串歷史記錄
# result = agent.run_sync('我們剛才討論了什麼？', message_history=long_history)

範例：根據執行上下文動態處理

History Processor 還可以接受一個可選的 RunContext 參數。這讓你可以取得當前執行的依賴項 (Dependencies) 或 Token 使用量等資訊，並據此做出判斷：

from pydantic_ai import Agent, ModelMessage, RunContext

def context_aware_processor(
    ctx: RunContext[None], 
    messages: list[ModelMessage]
) -> list[ModelMessage]:
    # 讀取當前的 Token 總使用量
    current_tokens = ctx.usage.total_tokens
    
    # 只有當 Token 使用量過高時，才進行訊息截斷
    if current_tokens and current_tokens > 1000:
        return messages[-3:]
    return messages

agent = Agent('openai:gpt-4o-mini', history_processors=[context_aware_processor])

透過靈活運用對話歷史的傳遞、JSON 序列化與動態的處理機制，你可以建構出具備長期記憶且成本可控的智慧型代理人應用。